From dba950443d2770778fef695215623f863ead0ba1 Mon Sep 17 00:00:00 2001 From: Wenchao An Date: Fri, 13 Nov 2026 23:01:38 +0800 Subject: [PATCH] feat(search): find conversations and jump to matching messages (#8032) * feat(search): find conversations and jump to matching messages * test(extensions): align example registry with search contribution * test(search): scope bookmark search to its extension * fix(search): address review correctness and scaling gaps --- backend/app/gateway/app.py | 2 + backend/app/gateway/routers/AGENTS.md | 3 + backend/app/gateway/routers/search.py | 160 ++++++++ backend/app/gateway/routers/threads.py | 3 + backend/app/gateway/search_corpus.py | 172 +++++++++ backend/docs/API.md | 32 ++ .../deerflow_extension_api/__init__.py | 2 +- .../deerflow_extension_api/contracts.py | 4 + .../deerflow_extension_api/search.py | 64 ++++ backend/packages/extension-api/pyproject.toml | 2 +- .../harness/deerflow/extensions/AGENTS.md | 2 + .../harness/deerflow/extensions/registry.py | 14 + .../deerflow/persistence/thread_meta/base.py | 1 + .../persistence/thread_meta/memory.py | 4 + .../deerflow/persistence/thread_meta/sql.py | 28 +- backend/packages/harness/pyproject.toml | 2 +- backend/scripts/run_replay_gateway.py | 5 + backend/tests/test_extension_api_contracts.py | 2 +- .../tests/test_thread_id_route_contract.py | 2 + backend/tests/test_workspace_search.py | 343 +++++++++++++++++ backend/tests/test_workspace_search_review.py | 123 ++++++ backend/uv.lock | 2 +- config.example.yaml | 9 + docs/guide/workspace.md | 11 + .../tests/test_plugin.py | 6 + .../README.md | 67 ++++ .../deerflow_workspace_search/__init__.py | 10 + .../deerflow_workspace_search/index.py | 227 +++++++++++ .../pyproject.toml | 16 + frontend/playwright.real-backend.config.ts | 2 +- .../[agent_name]/chats/[thread_id]/page.tsx | 17 +- frontend/src/components/workspace/AGENTS.md | 5 + .../components/workspace/chats/chat-page.tsx | 17 +- .../components/workspace/command-palette.tsx | 352 ++++++++++++++++-- .../workspace/search-message-window.tsx | 202 ++++++++++ .../workspace/workspace-nav-chat-list.tsx | 26 ++ frontend/src/core/extensions/catalog.ts | 5 + frontend/src/core/i18n/locales/en-US.ts | 38 ++ frontend/src/core/i18n/locales/types.ts | 32 ++ frontend/src/core/i18n/locales/zh-CN.ts | 35 ++ frontend/src/core/search/api.ts | 91 +++++ frontend/src/core/search/hooks.ts | 65 ++++ .../e2e-real-backend/workspace-search.spec.ts | 81 ++++ frontend/tests/e2e/bookmark-plugin.spec.ts | 9 +- .../tests/e2e/workspace-search-review.spec.ts | 191 ++++++++++ frontend/tests/e2e/workspace-search.spec.ts | 227 +++++++++++ .../tests/unit/core/search/hooks.dom.test.tsx | 58 +++ 47 files changed, 2722 insertions(+), 49 deletions(-) create mode 100644 backend/app/gateway/routers/search.py create mode 100644 backend/app/gateway/search_corpus.py create mode 100644 backend/packages/extension-api/deerflow_extension_api/search.py create mode 100644 backend/tests/test_workspace_search.py create mode 100644 backend/tests/test_workspace_search_review.py create mode 100644 examples/deerflow-extension-workspace-search/README.md create mode 100644 examples/deerflow-extension-workspace-search/deerflow_workspace_search/__init__.py create mode 100644 examples/deerflow-extension-workspace-search/deerflow_workspace_search/index.py create mode 100644 examples/deerflow-extension-workspace-search/pyproject.toml create mode 100644 frontend/src/components/workspace/search-message-window.tsx create mode 100644 frontend/src/core/search/api.ts create mode 100644 frontend/src/core/search/hooks.ts create mode 100644 frontend/tests/e2e-real-backend/workspace-search.spec.ts create mode 100644 frontend/tests/e2e/workspace-search-review.spec.ts create mode 100644 frontend/tests/e2e/workspace-search.spec.ts create mode 100644 frontend/tests/unit/core/search/hooks.dom.test.tsx diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 80a50a501f..ee1a2f5983 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -49,6 +49,7 @@ from app.gateway.routers import ( projects, runs, scheduled_tasks, + search, skill_mutations, skills, subagent_batches, @@ -1379,6 +1380,7 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for app.include_router(uploads.router) # Thread cleanup API is mounted at /api/threads/{thread_id} + app.include_router(search.router) app.include_router(threads.router) # Per-user activity feed (own prefix, so /api/threads/{thread_id} never captures it) diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md index 2c230be52d..80a3726f92 100644 --- a/backend/app/gateway/routers/AGENTS.md +++ b/backend/app/gateway/routers/AGENTS.md @@ -50,3 +50,6 @@ make test-blocking-io # after adding any file, config skill name, file name) is validated when it is created, including Windows reserved device names. Read and delete routes stay permissive so that data stored by an older release can still be exported and removed. +- **Search results use the latest owner-visible message.** A historical jump + resolves only the requested thread. Use the shared history seed path for + legacy checkpoints so it keeps its existing write reservation. diff --git a/backend/app/gateway/routers/search.py b/backend/app/gateway/routers/search.py new file mode 100644 index 0000000000..1472007f23 --- /dev/null +++ b/backend/app/gateway/routers/search.py @@ -0,0 +1,160 @@ +"""Workspace search and direct navigation to an owner-visible message.""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import logging +from dataclasses import asdict +from typing import Literal + +from fastapi import APIRouter, HTTPException, Query, Request +from pydantic import BaseModel, Field + +from app.gateway.authz import require_permission +from app.gateway.deps import get_run_event_store, get_run_manager, get_thread_store +from app.gateway.search_corpus import GatewaySearchCorpus +from deerflow.utils.thread_id import ThreadId + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api", tags=["search"]) + + +class WorkspaceSearchRequest(BaseModel): + query: str = Field(default="", max_length=256) + types: list[Literal["thread", "message"]] = Field(default_factory=lambda: ["thread", "message"], min_length=1, max_length=2) + archived: bool | None = Field(default=None, strict=True) + project_id: str | None = Field(default=None, max_length=64) + limit: int = Field(default=20, ge=1, le=100) + cursor: str | None = Field(default=None, max_length=4096) + + +def snippet(text: str, query: str) -> list[dict]: + """Plain text only. Map case-fold expansions back to source characters.""" + normalized = text.casefold() + position = normalized.find(query.casefold()) if query else -1 + if position < 0: + return [{"text": text[:200] + ("…" if len(text) > 200 else "")}] + offsets = [] + for index, character in enumerate(text): + offsets.extend([index] * len(character.casefold())) + if len(offsets) >= position + len(query.casefold()): + break + start, end = offsets[position], offsets[position + len(query.casefold()) - 1] + 1 + left, right = max(0, start - 65), min(len(text), end + 110) + return [{"text": ("…" if left else "") + text[left:start]}, {"text": text[start:end], "matched": True}, {"text": text[end:right] + ("…" if right < len(text) else "")}] + + +def _corpus(request: Request) -> GatewaySearchCorpus: + return GatewaySearchCorpus(get_thread_store(request), get_run_event_store(request), get_run_manager(request), str(request.state.auth.require_user().id), request=request) + + +def _encode_cursor(scope: str, phase: str, position) -> str: + return base64.urlsafe_b64encode(json.dumps([scope, phase, position]).encode()).decode() + + +@router.post("/search") +@require_permission("threads", "read") +async def search_workspace(body: WorkspaceSearchRequest, request: Request): + corpus = _corpus(request) + query = body.query.strip() + scope = hashlib.sha256(json.dumps([corpus.user_id, query, sorted(set(body.types)), body.archived, body.project_id, "project_id" in body.model_fields_set]).encode()).hexdigest() + phase, position = "thread", None + if body.cursor: + try: + saved_scope, phase, position = json.loads(base64.urlsafe_b64decode(body.cursor)) + if saved_scope != scope or phase not in {"thread", "message"}: + raise ValueError + if phase == "thread" and (not isinstance(position, list) or len(position) != 3 or not isinstance(position[0], int) or not all(isinstance(v, str) for v in position[1:])): + raise ValueError + if phase == "message" and position is not None and not isinstance(position, str): + raise ValueError + except (ValueError, TypeError) as exc: + raise HTTPException(400, "Invalid search cursor") from exc + all_threads = await corpus.threads() + threads = tuple(t for t in all_threads if (body.archived is None or t.archived == body.archived) and ("project_id" not in body.model_fields_set or t.project_id == body.project_id)) + lookup = {t.thread_id: t for t in threads} + providers = getattr(getattr(request.app.state, "extensions", None), "workspace_search_providers", ()) + messages_available = bool(providers) and request.state.auth.has_permission("runs", "read") + response = {"items": [], "cursor": None, "indexing": False, "messages_available": messages_available} + + def thread_hit(thread): + return { + "type": "thread", + "thread_id": thread.thread_id, + "title": thread.title, + "agent_name": thread.agent_name, + "archived": thread.archived, + "project_id": thread.project_id, + "updated_at": thread.updated_at, + "snippet": snippet(thread.title, query), + } + + if phase == "thread" and "thread" in body.types: + + def key(thread): + title = thread.title.casefold() + return (2 if query and title == query.casefold() else 1, thread.updated_at, thread.thread_id) + + matches = sorted((t for t in threads if not query or query.casefold() in t.title.casefold()), key=key, reverse=True) + if position is not None: + matches = [t for t in matches if key(t) < tuple(position)] + selected = matches[: body.limit] + response["items"].extend(thread_hit(t) for t in selected) + if len(matches) > body.limit: + response["cursor"] = _encode_cursor(scope, "thread", key(selected[-1])) + return response + if not query or "message" not in body.types or not messages_available: + return response + if len(response["items"]) == body.limit: + response["cursor"] = _encode_cursor(scope, "message", None) + return response + provider = providers[0][1] + provider_cursor = position if phase == "message" else None + # Invalidated candidates must not consume a visible result slot. Keep a + # bounded amount of work per HTTP page and expose continuation when needed. + for _ in range(5): + try: + page = await provider.search(corpus, query=query, threads=threads, limit=body.limit - len(response["items"]), cursor=provider_cursor) + except ValueError as exc: + raise HTTPException(400, "Invalid search cursor") from exc + except Exception: + logger.exception("Workspace search provider failed") + raise HTTPException(503, "Message search is temporarily unavailable") from None + response["indexing"] = page.indexing + for candidate in page.items: + thread = lookup.get(candidate.thread_id) + if thread is None or candidate.incarnation != thread.incarnation or not await corpus.owns(thread): + continue + # Re-read authoritative visible text before returning ANY snippet. + current = await corpus.latest_message(thread, candidate.message_id) + if current is None or current.seq != candidate.seq or query.casefold() not in current.text.casefold(): + continue + hit = thread_hit(thread) + hit.update(type="message", message_id=current.message_id, seq=current.seq, role=current.role, updated_at=current.created_at, snippet=snippet(current.text, query)) + response["items"].append(hit) + if page.indexing: + break + if page.cursor == provider_cursor and page.cursor is not None: + raise HTTPException(503, "Search pagination did not advance") + provider_cursor = page.cursor + if provider_cursor is None or len(response["items"]) >= body.limit: + break + if provider_cursor: + response["cursor"] = _encode_cursor(scope, "message", provider_cursor) + return response + + +@router.get("/threads/{thread_id}/messages/around") +@require_permission("runs", "read", owner_check=True, require_existing=True) +async def message_context(thread_id: ThreadId, request: Request, message_id: str = Query(min_length=1, max_length=512), seq: int | None = Query(default=None, ge=1), limit: int = Query(default=20, ge=1, le=50)): + corpus = _corpus(request) + window = await corpus.around(thread_id, message_id, seq=seq, limit=limit) + if window is None: + raise HTTPException(404, "Message is unavailable") + thread = await corpus.thread(thread_id) + if thread is None: + raise HTTPException(404, "Message is unavailable") + messages = [asdict(message) for row in window.pop("data") if (message := corpus.project(thread, row)) is not None] + return {**window, "messages": messages} diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index b468d96f83..16b79b2a59 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -530,6 +530,8 @@ class ThreadCreateRequest(BaseModel): class ThreadSearchRequest(BaseModel): """Request body for searching threads.""" + query: str | None = Field(default=None, max_length=256, description="Literal case-insensitive title substring") + archived: bool | None = Field(default=None, strict=True, description="Archive filter; omitted includes all, false includes legacy unarchived threads") metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata filter (exact match)") project_id: str | None = Field(default=None, description="Filter by project; explicit null = unassigned threads; omit key for all") @@ -1367,6 +1369,7 @@ async def search_threads(body: ThreadSearchRequest, request: Request) -> list[Th try: rows = await repo.search( metadata=body.metadata or None, + **({"query": body.query} if body.query is not None else {}), status=body.status, **({"archived": body.archived} if body.archived is not None else {}), project_id=project_filter, diff --git a/backend/app/gateway/search_corpus.py b/backend/app/gateway/search_corpus.py new file mode 100644 index 0000000000..3eaa1af7d9 --- /dev/null +++ b/backend/app/gateway/search_corpus.py @@ -0,0 +1,172 @@ +"""Permission-scoped transcript projection shared by search and message navigation.""" + +from __future__ import annotations + +import hashlib +import json +import re + +from deerflow_extension_api.search import SearchMessage, SearchMessagePage, SearchThread + +from app.gateway.conversation_access import _visible_text +from app.gateway.conversation_reader import default_history_hidden_run_ids, read_visible_message_page +from deerflow.runtime.events.message_identity import message_identity +from deerflow.utils.time import coerce_iso + + +def visible_search_text(row: dict, *, thread_id: str): + """Use the transcript reader and the chat UI's display-only exclusions.""" + content = row.get("content") + if isinstance(content, dict) and content.get("name") in {"summary", "loop_warning", "todo_reminder", "todo_completion_reminder"}: + return None + visible = _visible_text(row, thread_id=thread_id) + if visible and visible[0] == "user": + text = re.sub(r"<(current_uploads|uploaded_files|slash_skill_activation)>[\s\S]*?", "", visible[1]).strip() + return ("user", text) if text else None + return visible + + +class GatewaySearchCorpus: + def __init__(self, thread_store, event_store, run_manager, user_id: str, *, request=None): + self.thread_store = thread_store + self.event_store = event_store + self.run_manager = run_manager + self.user_id = user_id + self.request = request + + async def threads(self) -> tuple[SearchThread, ...]: + result = [] + offset = 0 + while True: + rows = await self.thread_store.search(user_id=self.user_id, limit=200, offset=offset) + for row in rows: + metadata = row.get("metadata") or {} + if metadata.get("deerflow_sidecar") is True: + continue + result.append(self.project_thread(row)) + offset += len(rows) + if len(rows) < 200: + return tuple(result) + + @staticmethod + def project_thread(row) -> SearchThread: + from app.gateway.services import custom_agent_name + + try: + agent_name = custom_agent_name(row.get("assistant_id")) + except (TypeError, ValueError): + agent_name = None + metadata = row.get("metadata") or {} + return SearchThread( + thread_id=row["thread_id"], + incarnation=str(row.get("incarnation") or row.get("created_at") or ""), + title=row.get("display_name") or "", + updated_at=coerce_iso(row.get("updated_at", "")), + agent_name=agent_name, + project_id=metadata.get("deerflow_project_id"), + archived=metadata.get("deerflow_archived") is True, + ) + + async def thread(self, thread_id: str) -> SearchThread | None: + row = await self.thread_store.get(thread_id, user_id=self.user_id) + if row is None or (row.get("metadata") or {}).get("deerflow_sidecar") is True: + return None + return self.project_thread(row) + + async def seed_legacy_history(self, thread: SearchThread) -> None: + if self.request is None or await self.event_store.list_messages(thread.thread_id, limit=1, user_id=None): + return + row = await self.thread_store.get(thread.thread_id, user_id=self.user_id) + if row is None or self.project_thread(row).incarnation != thread.incarnation: + return + from app.gateway.services import ensure_checkpoint_history_seeded + + await ensure_checkpoint_history_seeded(self.request, thread_id=thread.thread_id, assistant_id=row.get("assistant_id")) + + async def owns(self, thread: SearchThread) -> bool: + row = await self.thread_store.get(thread.thread_id, user_id=self.user_id) + return bool(row and str(row.get("incarnation") or row.get("created_at") or "") == thread.incarnation and (row.get("metadata") or {}).get("deerflow_sidecar") is not True) + + async def revision(self, thread: SearchThread) -> str | None: + if not await self.owns(thread): + return None + await self.seed_legacy_history(thread) + latest = await self.event_store.list_messages(thread.thread_id, limit=1, user_id=self.user_id) + hidden = await default_history_hidden_run_ids(self.run_manager, thread.thread_id, user_id=self.user_id) + revision = [thread.incarnation, thread.updated_at, latest[-1]["seq"] if latest else 0, sorted(hidden)] + return hashlib.sha256(json.dumps(revision).encode()).hexdigest() + + @staticmethod + def project(thread: SearchThread, row: dict) -> SearchMessage | None: + visible = visible_search_text(row, thread_id=thread.thread_id) + content = row.get("content") + identity = message_identity(content) if isinstance(content, dict) else None + if not visible or not identity: + return None + return SearchMessage(thread.thread_id, thread.incarnation, identity.removeprefix("message:"), row["run_id"], row["seq"], visible[0], visible[1], coerce_iso(row.get("created_at", ""))) + + async def messages(self, thread: SearchThread, *, before_seq: int | None) -> SearchMessagePage: + if not await self.owns(thread): + return SearchMessagePage() + rows, more = await self._page(thread.thread_id, limit=200, before_seq=before_seq) + items = tuple(message for row in rows if (message := self.project(thread, row)) is not None) + return SearchMessagePage(items, rows[0]["seq"] if more else None) + + async def _page(self, thread_id: str, *, limit: int, before_seq=None, after_seq=None): + return await read_visible_message_page( + event_store=self.event_store, + run_manager=self.run_manager, + thread_id=thread_id, + user_id=self.user_id, + limit=limit, + before_seq=before_seq, + after_seq=after_seq, + message_filter=lambda row: visible_search_text(row, thread_id=thread_id) is not None and message_identity(row.get("content", {})) is not None, + ) + + async def around(self, thread_id: str, message_id: str, *, limit: int = 20, seq: int | None = None) -> dict | None: + row = await self.thread_store.get(thread_id, user_id=self.user_id) + if row is None or (row.get("metadata") or {}).get("deerflow_sidecar") is True: + return None + await self.seed_legacy_history(self.project_thread(row)) + # Walk backward from the head so later versions win even when they + # live on another storage page. Sequence hints never select old text. + rows = await self.latest_rows(thread_id) + index = next((i for i, item in enumerate(rows) if message_identity(item["content"]) == f"message:{message_id}"), None) + if index is None or (seq is not None and rows[index]["seq"] != seq): + return None + target = rows[index] + seq = target["seq"] + left, right = max(0, index - limit), min(len(rows), index + limit + 1) + older, newer = rows[left:index], rows[index + 1 : right] + has_before, has_after = left > 0, right < len(rows) + # Recheck ownership/incarnation after the asynchronous history reads. + current = await self.thread_store.get(thread_id, user_id=self.user_id) + if current is None or current.get("incarnation") != row.get("incarnation"): + return None + return {"data": older + [target] + newer, "target_seq": seq, "has_before": has_before, "has_after": has_after} + + async def latest_rows(self, thread_id: str) -> list[dict]: + latest = {} + before = None + while True: + rows, more = await self._page(thread_id, limit=200, before_seq=before) + for row in reversed(rows): + latest.setdefault(message_identity(row["content"]), row) + if not more: + break + before = rows[0]["seq"] + return sorted(latest.values(), key=lambda row: row["seq"]) + + async def latest_message(self, thread: SearchThread, message_id: str) -> SearchMessage | None: + if not await self.owns(thread): + return None + before = None + while True: + rows, more = await self._page(thread.thread_id, limit=200, before_seq=before) + for row in reversed(rows): + if message_identity(row["content"]) == f"message:{message_id}": + return self.project(thread, row) + if not more: + return None + before = rows[0]["seq"] diff --git a/backend/docs/API.md b/backend/docs/API.md index 5af4a0a093..8b23eab12c 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -2283,6 +2283,38 @@ curl -X POST http://localhost:2026/api/langgraph/threads/abc123/runs/stream \ > section for details. Scheduled-task launches use > `scheduler.recursion_limit` from `config.yaml` instead of a client body. +## Workspace search + +`POST /api/threads/search` additionally accepts an optional `query` (up to 256 +characters). It matches a trimmed, case-insensitive literal substring of the +conversation title before pagination, preserving the existing array response. + +`POST /api/search` requires `threads:read` and returns grouped title/message +matches for the current user. Message results additionally require `runs:read` +and an installed `workspace_search` extension provider. + +```json +{"query":"报告","types":["thread","message"],"archived":null,"limit":20,"cursor":null} +``` + +`types` defaults to both kinds; `limit` is 1–100. `archived` is a nullable boolean; +`project_id` omitted includes every project, explicit null selects unassigned +threads. Blank queries return recent titles. The response is +`{items, cursor, indexing, messages_available}`. Each item carries `type`, +`thread_id`, `title`, `agent_name`, `archived`, `project_id`, `updated_at` and +`snippet` (plain-text `{text, matched?}` segments). Message hits also carry the +canonical `message_id`, thread-global `seq` and `role`. Preserve the query and +filters with each opaque cursor; invalid cursors return 400. When `indexing` is +true, retry the current page until historical backfill completes. Provider errors +return 503; without a provider, title search stays available. + +`GET /api/threads/{thread_id}/messages/around?message_id=...&seq=...&limit=20` +requires `runs:read` and thread ownership. `seq` is an optional exact version hint; +`limit` (1–50) bounds each side of the target. The response contains `messages`, +`target_seq`, `has_before` and `has_after`. Hidden/superseded targets and stale +version hints return 404. Earlier/later navigation uses the boundary message as +the next anchor. The UI keeps these windows separate from live history. + ## Chat archive and restore `POST /api/threads/search` accepts `archived: true` for archived chats or diff --git a/backend/packages/extension-api/deerflow_extension_api/__init__.py b/backend/packages/extension-api/deerflow_extension_api/__init__.py index f2b5e16cfb..dc5f6d4820 100644 --- a/backend/packages/extension-api/deerflow_extension_api/__init__.py +++ b/backend/packages/extension-api/deerflow_extension_api/__init__.py @@ -102,7 +102,7 @@ from deerflow_extension_api.state import ExtensionData #: Contract version. Before 1.0, minors may break and patches are additive. #: From 1.0 on, bump the major for breaking changes. -API_VERSION = "0.2.11" +API_VERSION = "0.2.12" __all__ = [ "AGENT_RUNS_CONTEXT_KEY", diff --git a/backend/packages/extension-api/deerflow_extension_api/contracts.py b/backend/packages/extension-api/deerflow_extension_api/contracts.py index 47090864b7..b139e11294 100644 --- a/backend/packages/extension-api/deerflow_extension_api/contracts.py +++ b/backend/packages/extension-api/deerflow_extension_api/contracts.py @@ -210,6 +210,10 @@ class ExtensionRegistry(Protocol): """ return False + def workspace_search(self, provider: Any) -> bool: + """Register one WorkspaceSearchProvider; False on hosts without search.""" + return False + def middlewares(self, contributor: MiddlewareContributor) -> None: return None diff --git a/backend/packages/extension-api/deerflow_extension_api/search.py b/backend/packages/extension-api/deerflow_extension_api/search.py new file mode 100644 index 0000000000..d8953d0fa7 --- /dev/null +++ b/backend/packages/extension-api/deerflow_extension_api/search.py @@ -0,0 +1,64 @@ +"""Owner-bound workspace search contributions. No framework or storage dependency.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + + +@dataclass(frozen=True) +class SearchThread: + thread_id: str + incarnation: str + title: str + updated_at: str + agent_name: str | None = None + project_id: str | None = None + archived: bool = False + + +@dataclass(frozen=True) +class SearchMessage: + thread_id: str + incarnation: str + message_id: str + run_id: str + seq: int + role: str + text: str + created_at: str + + +@dataclass(frozen=True) +class SearchMessagePage: + items: tuple[SearchMessage, ...] = () + before_seq: int | None = None + + +class SearchCorpus(Protocol): + """Host-scoped visible conversation reader; IDs never grant access.""" + + user_id: str + + async def threads(self) -> tuple[SearchThread, ...]: ... + + async def revision(self, thread: SearchThread) -> str | None: ... + + async def messages(self, thread: SearchThread, *, before_seq: int | None) -> SearchMessagePage: ... + + +@dataclass(frozen=True) +class SearchPage: + items: tuple[SearchMessage, ...] = () + cursor: str | None = None + indexing: bool = False + + +class WorkspaceSearchProvider(Protocol): + """Return literal, case-insensitive matches. Host rechecks each candidate. + + Cursor is provider-owned, opaque and scoped to the query. Services own index + resources and stop background work before host storage closes. + """ + + async def search(self, corpus: SearchCorpus, *, query: str, threads: tuple[SearchThread, ...], limit: int, cursor: str | None = None) -> SearchPage: ... diff --git a/backend/packages/extension-api/pyproject.toml b/backend/packages/extension-api/pyproject.toml index 2a46b18122..713276e1ed 100644 --- a/backend/packages/extension-api/pyproject.toml +++ b/backend/packages/extension-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deerflow-extension-api" -version = "0.2.11" +version = "0.2.12" description = "Public contracts for DeerFlow extensions" requires-python = ">=3.12" # Keep the contract package import-light and independent from the host. Public diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md index c4a385017d..113c652ac2 100644 --- a/backend/packages/harness/deerflow/extensions/AGENTS.md +++ b/backend/packages/harness/deerflow/extensions/AGENTS.md @@ -86,6 +86,8 @@ hook directory, so they never run a developer's or CI's Git hooks. look identical. - **Shutdown stops accepting detached observations before the memory flush**, so an accepted observation is never registered after shutdown began. +- **Search indexes are rebuildable.** The Gateway rechecks each indexed hit + against the current owner-visible message before returning text. ## Full-stack plugins and host access diff --git a/backend/packages/harness/deerflow/extensions/registry.py b/backend/packages/harness/deerflow/extensions/registry.py index 3749c4970d..4728569bd8 100644 --- a/backend/packages/harness/deerflow/extensions/registry.py +++ b/backend/packages/harness/deerflow/extensions/registry.py @@ -57,6 +57,7 @@ class LoadedExtensions: routers: tuple[tuple[str, Any], ...] = () plugins: tuple[tuple[str, PluginContribution], ...] = () knowledge_providers: tuple[tuple[str, tuple[str, Any]], ...] = () + workspace_search_providers: tuple[tuple[str, Any], ...] = () documentation: tuple[tuple[str, dict], ...] = () # Precomputed attributes, not methods: hook sites read one attribute to @@ -87,6 +88,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._routers: list[_Entry] = [] self._plugins: list[_Entry] = [] self._knowledge_providers: list[_Entry] = [] + self._workspace_search_providers: list[_Entry] = [] self._current_source: str | None = None self._model_access: ModelInvocationScope | None = None @@ -149,6 +151,14 @@ class ExtensionRegistry(ExtensionRegistryContract): self._plugins.append((self._source(), contribution)) return True + def workspace_search(self, provider: Any) -> bool: + if self._workspace_search_providers: + raise ValueError("Only one workspace search provider may be installed") + if not inspect.iscoroutinefunction(getattr(provider, "search", None)): + raise ValueError("Workspace search requires an async search method") + self._workspace_search_providers.append((self._source(), provider)) + return True + def middlewares(self, contributor: MiddlewareContributor) -> None: self._middlewares.append((self._source(), contributor)) @@ -207,6 +217,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._routers, self._plugins, self._knowledge_providers, + self._workspace_search_providers, ): bucket[:] = [entry for entry in bucket if entry[0] != source] @@ -222,6 +233,7 @@ class ExtensionRegistry(ExtensionRegistryContract): len(self._routers), len(self._plugins), len(self._knowledge_providers), + len(self._workspace_search_providers), ) def rollback_to(self, mark: tuple[int, ...]) -> None: @@ -242,6 +254,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._routers, self._plugins, self._knowledge_providers, + self._workspace_search_providers, ), mark, strict=True, @@ -260,6 +273,7 @@ class ExtensionRegistry(ExtensionRegistryContract): routers=tuple(self._routers), plugins=tuple(self._plugins), knowledge_providers=tuple(self._knowledge_providers), + workspace_search_providers=tuple(self._workspace_search_providers), has_middleware_contributors=bool(self._middlewares), has_task_lifecycle=bool(self._task_lifecycle), has_system_model_observers=bool(self._system_model_observers), diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/base.py b/backend/packages/harness/deerflow/persistence/thread_meta/base.py index 6216043558..cc68650fc1 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/base.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/base.py @@ -97,6 +97,7 @@ class ThreadMetaStore(abc.ABC): self, *, metadata: dict[str, Any] | None = None, + query: str | None = None, status: str | None = None, archived: bool | None = None, project_id: str | None | _ProjectFilterUnset = PROJECT_FILTER_UNSET, diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py index c1913932f8..a79da62ba6 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py @@ -106,6 +106,7 @@ class MemoryThreadMetaStore(ThreadMetaStore): self, *, metadata: dict[str, Any] | None = None, + query: str | None = None, status: str | None = None, archived: bool | None = None, project_id: str | None | _ProjectFilterUnset = PROJECT_FILTER_UNSET, @@ -150,6 +151,9 @@ class MemoryThreadMetaStore(ThreadMetaStore): records = [record for record in records if isinstance(record.get("metadata"), dict) and all(json_value_matches(record["metadata"], key, value) for key, value in metadata.items())] if archived is not None: records = [record for record in records if ((record.get("metadata") or {}).get(THREAD_ARCHIVED_METADATA_KEY) is True) == archived] + if query and query.strip(): + needle = query.strip().casefold() + records = [record for record in records if needle in (record.get("display_name") or "").casefold()] records.sort(key=self._sort_key, reverse=True) return records[offset : offset + limit] diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py index 3a4fc16da3..013b0c74f8 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py @@ -215,6 +215,7 @@ class ThreadMetaRepository(ThreadMetaStore): self, *, metadata: dict[str, Any] | None = None, + query: str | None = None, status: str | None = None, archived: bool | None = None, project_id: str | None | _ProjectFilterUnset = PROJECT_FILTER_UNSET, @@ -268,10 +269,31 @@ class ThreadMetaRepository(ThreadMetaStore): else: stmt = stmt.where(ThreadMetaRow.project_id == project_id) - stmt = stmt.limit(limit).offset(offset) + needle = query.strip().casefold() if query else "" async with self._sf() as session: - result = await session.execute(stmt) - return [self._row_to_dict(r) for r in result.scalars()] + if not needle: + result = await session.execute(stmt.limit(limit).offset(offset)) + return [self._row_to_dict(r) for r in result.scalars()] + # Casefold is dialect-independent (including ß -> ss). Scan bounded + # metadata batches and stop as soon as the requested page is full. + matches = [] + scanned, skipped = 0, 0 + while len(matches) < limit: + result = await session.execute(stmt.limit(200).offset(scanned)) + batch = list(result.scalars()) + for row in batch: + if needle not in (row.display_name or "").casefold(): + continue + if skipped < offset: + skipped += 1 + else: + matches.append(self._row_to_dict(row)) + if len(matches) == limit: + break + if len(batch) < 200: + break + scanned += len(batch) + return matches 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).""" diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 03c8e054c3..dd03e1e7ef 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ # the contract version it implements, extensions declare ranges. A range # here would let pip resolve a newer contract package than this harness # implements, making newer extensions look supported at runtime. - "deerflow-extension-api==0.2.11", + "deerflow-extension-api==0.2.12", "dotenv>=0.9.9", "exa-py>=1.0.0", "httpx>=0.28.0", diff --git a/backend/scripts/run_replay_gateway.py b/backend/scripts/run_replay_gateway.py index 996ee4d422..a68e473f71 100644 --- a/backend/scripts/run_replay_gateway.py +++ b/backend/scripts/run_replay_gateway.py @@ -29,6 +29,7 @@ def main() -> int: parser.add_argument("--port", type=int, default=8011) parser.add_argument("--fixture", default=str(_BACKEND / "tests" / "fixtures" / "replay" / "write_read_file.ultra.json")) parser.add_argument("--cors", default="http://localhost:3000") + parser.add_argument("--workspace-search", action="store_true", help="Enable the packaged workspace search provider for full-stack search tests") args = parser.parse_args() from _replay_fixture import REPLAY_MODEL_BLOCK, build_config_yaml, prepare_hermetic_extras @@ -36,6 +37,10 @@ def main() -> int: home = Path(tempfile.mkdtemp(prefix="replay-gw-")) cfg = home / "config.yaml" cfg.write_text(build_config_yaml(model_block=REPLAY_MODEL_BLOCK, home=home), encoding="utf-8") + if args.workspace_search: + sys.path.insert(0, str(_BACKEND.parent / "examples/deerflow-extension-workspace-search")) + with cfg.open("a", encoding="utf-8") as stream: + stream.write("\nplugins:\n - name: workspace-search\n use: deerflow_workspace_search:install\n table_prefix: extension_workspace_search_\n required: true\n") # Override (not setdefault): the replay gateway must be hermetic, so an outer # DEER_FLOW_HOME can't leak in and shift prompt-affecting paths/skills. diff --git a/backend/tests/test_extension_api_contracts.py b/backend/tests/test_extension_api_contracts.py index f8dfb036d0..9e7f3e892e 100644 --- a/backend/tests/test_extension_api_contracts.py +++ b/backend/tests/test_extension_api_contracts.py @@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package(): """Every additive contract slice bumps both gates together.""" from importlib.metadata import version - assert API_VERSION == "0.2.11" + assert API_VERSION == "0.2.12" assert API_VERSION == version("deerflow-extension-api") diff --git a/backend/tests/test_thread_id_route_contract.py b/backend/tests/test_thread_id_route_contract.py index 368ff8344a..4dd716135c 100644 --- a/backend/tests/test_thread_id_route_contract.py +++ b/backend/tests/test_thread_id_route_contract.py @@ -77,6 +77,7 @@ def _collect_thread_id_routes(): project_documents, runs, scheduled_tasks, + search, skills, subagent_batches, suggestions, @@ -93,6 +94,7 @@ def _collect_thread_id_routes(): project_documents, runs, scheduled_tasks, + search, skills, subagent_batches, suggestions, diff --git a/backend/tests/test_workspace_search.py b/backend/tests/test_workspace_search.py new file mode 100644 index 0000000000..e79c6360fd --- /dev/null +++ b/backend/tests/test_workspace_search.py @@ -0,0 +1,343 @@ +"""Search must find the visible transcript, with the same owner and identity rules.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +from langgraph.store.memory import InMemoryStore +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.persistence.thread_meta.model import ThreadMetaRow +from deerflow.persistence.thread_meta.sql import ThreadMetaRepository +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.manager import EditReplayVisibility + + +@pytest.fixture(params=["memory", "sqlite"]) +async def threads(request, tmp_path): + if request.param == "memory": + yield MemoryThreadMetaStore(InMemoryStore()) + return + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'threads.db'}") + async with engine.begin() as connection: + await connection.run_sync(ThreadMetaRow.metadata.create_all) + yield ThreadMetaRepository(async_sessionmaker(engine, expire_on_commit=False)) + await engine.dispose() + + +@pytest.mark.anyio +async def test_title_query_is_literal_unicode_and_precedes_pagination(threads): + for thread_id, title in [("a", "旧调研报告"), ("b", "新调研报告"), ("c", "无关标题"), ("d", "Straße_100%")]: + await threads.create(thread_id, display_name=title, user_id="alice") + await threads.create("foreign", display_name="调研报告", user_id="bob") + assert [r["thread_id"] for r in await threads.search(query="报告", user_id="alice", limit=1, offset=1)] == ["a"] + assert [r["thread_id"] for r in await threads.search(query="STRASSE_100%", user_id="alice")] == ["d"] + assert [r["thread_id"] for r in await threads.search(query="%", user_id="alice")] == ["d"] + assert len(await threads.search(query=" ", user_id="alice")) == 4 + + +@pytest.mark.anyio +async def test_search_corpus_hides_internal_and_superseded_text_and_normalizes_ids(threads): + from app.gateway.search_corpus import GatewaySearchCorpus + + events = MemoryRunEventStore() + manager = AsyncMock() + manager.list_successful_regenerate_sources.return_value = {"old"} + manager.list_edit_replay_visibility.return_value = EditReplayVisibility() + await threads.create("t", user_id="alice", display_name="报告") + for run, mid, kind, extra, caller, body in [ + ("r", "u__user", "human", {"original_user_content": "原始报告"}, "lead_agent", "注入的提示"), + ("r", "a", "ai", {}, "lead_agent", "回答报告"), + ("r", "hidden", "ai", {"hide_from_ui": True}, "lead_agent", "隐藏报告"), + ("old", "old", "ai", {}, "lead_agent", "旧版本报告"), + ("r", "sub", "ai", {}, "subagent:research", "子任务报告"), + ("r", "tool", "tool", {}, "lead_agent", "工具日志报告"), + ]: + await events.put(thread_id="t", run_id=run, event_type="message", category="message", content={"id": mid, "type": kind, "content": body, "additional_kwargs": extra}, metadata={"caller": caller}) + corpus = GatewaySearchCorpus(threads, events, manager, "alice") + thread = (await corpus.threads())[0] + page = await corpus.messages(thread, before_seq=None) + assert [(m.message_id, m.text) for m in page.items] == [("u", "原始报告"), ("a", "回答报告")] + assert await GatewaySearchCorpus(threads, events, manager, "bob").threads() == () + assert (await GatewaySearchCorpus(threads, events, manager, "bob").messages(thread, before_seq=None)).items == () + + +@pytest.mark.anyio +async def test_message_window_finds_unloaded_history_and_rejects_hidden_target(threads): + from app.gateway.search_corpus import GatewaySearchCorpus + + events = MemoryRunEventStore() + manager = AsyncMock() + manager.list_successful_regenerate_sources.return_value = set() + manager.list_edit_replay_visibility.return_value = EditReplayVisibility() + await threads.create("t", user_id="alice") + for i in range(140): + await events.put(thread_id="t", run_id="r", event_type="message", category="message", content={"id": f"m-{i}", "type": "human" if i % 2 == 0 else "ai", "content": f"报告 {i}"}) + corpus = GatewaySearchCorpus(threads, events, manager, "alice") + window = await corpus.around("t", "m-4", limit=3) + assert window["target_seq"] == 5 + assert [r["seq"] for r in window["data"]] == list(range(2, 9)) + assert window["has_before"] and window["has_after"] + manager.list_successful_regenerate_sources.return_value = {"r"} + assert await corpus.around("t", "m-4", limit=3) is None + assert await GatewaySearchCorpus(threads, events, manager, "bob").around("t", "m-4", limit=3) is None + + +@pytest.fixture +async def index(tmp_path, monkeypatch): + from pathlib import Path + + from deerflow_extension_api import ExtensionRuntimeDeps + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-workspace-search")) + from deerflow_workspace_search.index import WorkspaceSearchIndex + + instance = WorkspaceSearchIndex(path=str(tmp_path / "search.db"), wait_seconds=10) + await instance.start(ExtensionRuntimeDeps()) + yield instance + await instance.stop() + + +def manager(): + result = AsyncMock() + result.list_successful_regenerate_sources.return_value = set() + result.list_edit_replay_visibility.return_value = EditReplayVisibility() + return result + + +async def put(events, thread_id, message_id, body, *, run_id="r", extra=None): + return await events.put(thread_id=thread_id, run_id=run_id, event_type="llm.ai.response", category="message", content={"id": message_id, "type": "ai", "content": body, "additional_kwargs": extra or {}}) + + +@pytest.mark.anyio +@pytest.mark.parametrize("event_backend", ["memory", "jsonl", "sqlite"]) +async def test_index_backfills_short_chinese_code_unicode_and_literal_punctuation(index, tmp_path, event_backend): + from app.gateway.search_corpus import GatewaySearchCorpus + from deerflow.persistence.models.run_event import RunEventRow + from deerflow.runtime.events.store.db import DbRunEventStore + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + from deerflow.runtime.user_context import get_effective_user_id + + owner = str(get_effective_user_id()) + threads = MemoryThreadMetaStore(InMemoryStore()) + await threads.create("t", user_id=owner) + engine = None + if event_backend == "sqlite": + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'events.db'}") + async with engine.begin() as connection: + await connection.run_sync(RunEventRow.metadata.create_all) + events = DbRunEventStore(async_sessionmaker(engine, expire_on_commit=False)) + else: + events = MemoryRunEventStore() if event_backend == "memory" else JsonlRunEventStore(tmp_path / "events") + try: + for i in range(230): + await put(events, "t", f"m-{i}", f"普通历史 {i}") + await put(events, "t", "match", "旧版调研报告") + await put(events, "t", "match", "调研报告 Straße parse_config_v2 100%_done ") + await put(events, "t", "hidden", "机密报告", extra={"hide_from_ui": True}) + corpus = GatewaySearchCorpus(threads, events, manager(), owner) + scope = await corpus.threads() + for query in ["报告", "调研报告", "STRASSE", "parse_config_v2", "%_", "100%_done"]: + page = await index.search(corpus, query=query, threads=scope, limit=10) + assert not page.indexing + assert [m.message_id for m in page.items] == ["match"] + assert "旧版" not in page.items[0].text + for query in ["机密", '" OR *', "100X_done"]: + assert not (await index.search(corpus, query=query, threads=scope, limit=10)).items + finally: + if engine: + await engine.dispose() + + +@pytest.mark.anyio +async def test_index_refreshes_regeneration_deletion_recreation_and_owner_scope(index): + from app.gateway.search_corpus import GatewaySearchCorpus + + threads = MemoryThreadMetaStore(InMemoryStore()) + events = MemoryRunEventStore() + runs = manager() + await threads.create("t", user_id="alice") + await threads.create("other", user_id="bob") + await put(events, "t", "first", "报告第一版", run_id="old") + await put(events, "other", "foreign", "报告秘密") + corpus = GatewaySearchCorpus(threads, events, runs, "alice") + + async def search(query="报告", limit=10, cursor=None): + return await index.search(corpus, query=query, threads=await corpus.threads(), limit=limit, cursor=cursor) + + assert [m.message_id for m in (await search()).items] == ["first"] + await put(events, "t", "second", "报告第二版", run_id="new") + runs.list_successful_regenerate_sources.return_value = {"old"} + assert [m.message_id for m in (await search()).items] == ["second"] + await threads.delete("t", user_id="alice") + assert not (await search()).items + await events.delete_by_thread("t", user_id="alice") + await threads.create("t", user_id="alice") + assert not (await search()).items + + +@pytest.mark.anyio +async def test_index_cursor_pages_every_match_and_rejects_another_query(index): + from app.gateway.search_corpus import GatewaySearchCorpus + + threads = MemoryThreadMetaStore(InMemoryStore()) + await threads.create("t", user_id="alice") + events = MemoryRunEventStore() + for i in range(7): + await put(events, "t", f"m-{i}", "报告") + corpus = GatewaySearchCorpus(threads, events, manager(), "alice") + scope = await corpus.threads() + first = await index.search(corpus, query="报告", threads=scope, limit=2) + assert first.cursor + with pytest.raises(ValueError): + await index.search(corpus, query="other", threads=scope, limit=2, cursor=first.cursor) + ids = [m.message_id for m in first.items] + page = first + while page.cursor: + page = await index.search(corpus, query="报告", threads=scope, limit=2, cursor=page.cursor) + ids.extend(m.message_id for m in page.items) + assert ids == [f"m-{i}" for i in reversed(range(7))] + + +@pytest.mark.anyio +async def test_http_search_and_jump_are_owner_bound_and_keep_title_contract(index): + from types import SimpleNamespace + from uuid import UUID + + from _router_auth_helpers import make_authed_test_app + from httpx import ASGITransport, AsyncClient + + from app.gateway.auth.models import User + from app.gateway.routers import search + from app.gateway.routers import threads as thread_routes + + user = User(id=UUID("00000000-0000-0000-0000-000000000123"), email="search@example.com", password_hash="unused") + app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True) + app.include_router(search.router) + app.include_router(thread_routes.router) + app.state.thread_store = MemoryThreadMetaStore(InMemoryStore()) + app.state.run_event_store = MemoryRunEventStore() + app.state.run_manager = manager() + app.state.extensions = SimpleNamespace(workspace_search_providers=(("test", index),)) + await app.state.thread_store.create("t", user_id=str(user.id), display_name="调研报告", assistant_id="researcher") + await app.state.thread_store.create("other", user_id="bob", display_name="调研报告秘密") + await put(app.state.run_event_store, "t", "old-message", "包含报告正文") + await put(app.state.run_event_store, "other", "secret", "报告秘密") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + title = await client.post("/api/threads/search", json={"query": "报告"}) + assert title.status_code == 200 + assert [r["thread_id"] for r in title.json()] == ["t"] + response = await client.post("/api/search", json={"query": "报告"}) + assert response.status_code == 200, response.text + hits = response.json()["items"] + assert [h["type"] for h in hits] == ["thread", "message"] + assert all(h["agent_name"] == "researcher" for h in hits) + assert hits[1]["snippet"][1] == {"text": "报告", "matched": True} + jump = await client.get("/api/threads/t/messages/around", params={"message_id": "old-message", "seq": 1}) + assert jump.status_code == 200, jump.text + assert jump.json()["messages"][0]["text"] == "包含报告正文" + assert (await client.get("/api/threads/other/messages/around", params={"message_id": "secret"})).status_code in {403, 404} + assert (await client.post("/api/search", json={"query": "报告", "cursor": "invalid"})).status_code == 400 + await app.state.thread_store.delete("t", user_id=str(user.id)) + assert not (await client.post("/api/search", json={"query": "报告"})).json()["items"] + + +def test_provider_registration_rolls_back_and_rejects_duplicates(): + from deerflow.extensions.registry import ExtensionRegistry + + registry = ExtensionRegistry() + provider = AsyncMock() + before = registry.mark() + with registry.attributed_to("search"): + assert registry.workspace_search(provider) + with pytest.raises(ValueError): + registry.workspace_search(provider) + assert registry.build().workspace_search_providers == (("search", provider),) + registry.rollback_to(before) + assert registry.build().workspace_search_providers == () + + +def test_snippet_preserves_unicode_expansions_and_never_returns_markup(): + from app.gateway.routers.search import snippet + + parts = snippet("前缀 Straße ", "STRASSE") + assert parts[1] == {"text": "Straße", "matched": True} + assert "".join(p["text"] for p in parts) == "前缀 Straße " + + +@pytest.mark.anyio +async def test_search_projection_matches_chat_control_message_exclusions(index): + from app.gateway.search_corpus import GatewaySearchCorpus + + threads = MemoryThreadMetaStore(InMemoryStore()) + events = MemoryRunEventStore() + await threads.create("t", user_id="alice") + for name in ("summary", "loop_warning", "todo_reminder", "todo_completion_reminder"): + await events.put(thread_id="t", run_id="r", event_type="llm.ai.response", category="message", content={"type": "ai", "id": name, "name": name, "content": "hidden marker"}) + await events.put(thread_id="t", run_id="r", event_type="llm.human.input", category="message", content={"type": "human", "id": "activation", "content": "hidden marker"}) + await events.put(thread_id="t", run_id="r", event_type="llm.human.input", category="message", content={"type": "human", "id": "upload", "content": "user texthidden marker"}) + corpus = GatewaySearchCorpus(threads, events, manager(), "alice") + scope = await corpus.threads() + assert not (await index.search(corpus, query="hidden marker", threads=scope, limit=20)).items + assert (await index.search(corpus, query="user text", threads=scope, limit=20)).items[0].text == "user text" + + +@pytest.mark.anyio +async def test_slow_backfill_is_reported_and_reused(index, monkeypatch): + import asyncio + from types import SimpleNamespace + + completed = asyncio.Event() + refresh = AsyncMock(side_effect=lambda _: completed.wait()) + + async def wait_for_release(_): + await completed.wait() + return True + + refresh.side_effect = wait_for_release + monkeypatch.setattr(index, "_refresh", refresh) + index.wait_seconds = 0.001 + corpus = SimpleNamespace(user_id="alice") + pending = await index.search(corpus, query="报告", threads=(), limit=10) + assert pending.indexing and not pending.items + completed.set() + index.wait_seconds = 10 + ready = await index.search(corpus, query="报告", threads=(), limit=10) + assert not ready.indexing + assert refresh.await_count == 1 + + +@pytest.mark.anyio +async def test_route_filters_and_permission_boundary(monkeypatch, index): + from types import SimpleNamespace + + import _router_auth_helpers as auth_helpers + from httpx import ASGITransport, AsyncClient + + from app.gateway.auth.models import User + from app.gateway.routers import search + + user = User(email="filters@example.com") + app = auth_helpers.make_authed_test_app(user_factory=lambda: user, bind_current_user=True) + app.include_router(search.router) + app.state.thread_store = MemoryThreadMetaStore(InMemoryStore()) + app.state.run_event_store = MemoryRunEventStore() + app.state.run_manager = manager() + app.state.extensions = SimpleNamespace(workspace_search_providers=(("test", index),)) + for tid, meta in (("active", {}), ("archived", {"deerflow_archived": True}), ("project", {"deerflow_project_id": "p"}), ("sidecar", {"deerflow_sidecar": True})): + await app.state.thread_store.create(tid, user_id=str(user.id), display_name="报告", metadata=meta) + await put(app.state.run_event_store, tid, "m", "报告正文") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post("/api/search", json={"query": "报告", "archived": False, "project_id": None}) + assert response.status_code == 200, response.text + assert [h["thread_id"] for h in response.json()["items"]] == ["active", "active"] + monkeypatch.setattr(auth_helpers, "_STUB_PERMISSIONS", ["threads:read"]) + response = await client.post("/api/search", json={"query": "报告", "project_id": "p"}) + assert response.status_code == 200 + assert not response.json()["messages_available"] + assert [h["type"] for h in response.json()["items"]] == ["thread"] + assert (await client.get("/api/threads/project/messages/around?message_id=m")).status_code == 403 + monkeypatch.setattr(auth_helpers, "_STUB_PERMISSIONS", []) + assert (await client.post("/api/search", json={"query": "报告"})).status_code == 403 diff --git a/backend/tests/test_workspace_search_review.py b/backend/tests/test_workspace_search_review.py new file mode 100644 index 0000000000..76541780ed --- /dev/null +++ b/backend/tests/test_workspace_search_review.py @@ -0,0 +1,123 @@ +"""Review regressions use real message stores and the SQL index.""" + +import inspect +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from langgraph.store.memory import InMemoryStore +from sqlalchemy import event +from test_workspace_search import index, manager, put # noqa: F401 + +from app.gateway.search_corpus import GatewaySearchCorpus +from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.runtime.events.store.memory import MemoryRunEventStore + + +@pytest.mark.anyio +async def test_latest_versions_cross_pages_and_single_thread_lookup(monkeypatch): + from app.gateway.routers import search + + threads = MemoryThreadMetaStore(InMemoryStore()) + events = MemoryRunEventStore() + await threads.create("t", user_id="alice") + await put(events, "t", "same", "obsolete needle") + for number in range(220): + await put(events, "t", f"middle-{number}", "context") + await put(events, "t", "same", "current answer") + corpus = GatewaySearchCorpus(threads, events, manager(), "alice") + thread = await corpus.thread("t") + assert (await corpus.latest_message(thread, "same")).text == "current answer" + assert await corpus.around("t", "same", seq=1) is None + window = await corpus.around("t", "same", seq=222, limit=50) + assert len({row["content"]["id"] for row in window["data"]}) == len(window["data"]) + assert "obsolete needle" not in str(window) + threads.search = AsyncMock(side_effect=AssertionError("jump must not enumerate other threads")) + monkeypatch.setattr(search, "_corpus", lambda request: corpus) + result = await inspect.unwrap(search.message_context)("t", SimpleNamespace(), message_id="same", seq=222, limit=5) + assert result["target_seq"] == 222 + threads.search.assert_not_called() + + +@pytest.mark.anyio +async def test_provider_bounds_membership_parameters(index): # noqa: F811 + from deerflow_extension_api.search import SearchMessage, SearchThread + + threads = tuple(SearchThread(f"t-{i}", "v1", "Title", "2026-01-01") for i in range(1200)) + await index._replace("alice", threads[-1], "revision", [SearchMessage(threads[-1].thread_id, "v1", "message", "run", 1, "assistant", "needle", "2026-01-01")]) + index._refresh = AsyncMock(return_value=True) + counts = [] + + def record(connection, cursor, statement, parameters, context, executemany): + if "extension_workspace_search_documents" in statement: + counts.append(len(parameters)) + + event.listen(index._engine.sync_engine, "before_cursor_execute", record) + try: + result = await index.search(SimpleNamespace(user_id="alice"), query="needle", threads=threads, limit=20) + assert [item.thread_id for item in result.items] == [threads[-1].thread_id] + assert max(counts) < 300 + finally: + event.remove(index._engine.sync_engine, "before_cursor_execute", record) + + +@pytest.mark.anyio +async def test_fts_startup_skips_clean_rebuild_and_repairs_fallback_writes(index): # noqa: F811 + from deerflow_extension_api import ExtensionRuntimeDeps + from deerflow_extension_api.search import SearchMessage, SearchThread + + if not index._fts: + pytest.skip("SQLite has no trigram FTS") + thread = SearchThread("t", "v1", "Title", "2026-01-01") + + def message(body): + return SearchMessage("t", "v1", "message", "run", 1, "assistant", body, "2026-01-01") + + await index._replace("alice", thread, "one", [message("old needle")]) + await index.stop() + original = index.__class__(path=index.path) + await original.start(ExtensionRuntimeDeps()) + original._owns_engine = False + sf = original._sf + statements = [] + event.listen(original._engine.sync_engine, "before_cursor_execute", lambda c, cu, sql, p, ctx, many: statements.append(sql)) + try: + await original.start(ExtensionRuntimeDeps(session_factory=sf)) + assert not any(sql.strip() == "DELETE FROM extension_workspace_search_fts" for sql in statements) + original._fts = False + await original._replace("alice", thread, "two", [message("fresh needle")]) + statements.clear() + await original.start(ExtensionRuntimeDeps(session_factory=sf)) + assert any(sql.strip() == "DELETE FROM extension_workspace_search_fts" for sql in statements) + original._refresh = AsyncMock(return_value=True) + result = await original.search(SimpleNamespace(user_id="alice"), query="fresh", threads=(thread,), limit=20) + assert [item.text for item in result.items] == ["fresh needle"] + finally: + await original.stop() + await original._engine.dispose() + + +@pytest.mark.anyio +async def test_legacy_checkpoint_is_seeded_before_revision_completion(monkeypatch): + from langchain_core.messages import HumanMessage + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.graph import END, START, MessagesState, StateGraph + + from app.gateway import services + + saver = InMemorySaver() + builder = StateGraph(MessagesState) + builder.add_edge(START, END) + graph = builder.compile(checkpointer=saver) + config = {"configurable": {"thread_id": "legacy"}} + await graph.ainvoke({"messages": [HumanMessage(content="legacy needle", id="old")]}, config) + threads = MemoryThreadMetaStore(InMemoryStore()) + await threads.create("legacy", user_id="alice") + events = MemoryRunEventStore() + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(run_event_store=events, checkpointer=saver))) + monkeypatch.setattr(services, "abuild_checkpoint_state_accessor", AsyncMock(return_value=(SimpleNamespace(aget=graph.aget_state), config))) + corpus = GatewaySearchCorpus(threads, events, manager(), "alice", request=request) + thread = await corpus.thread("legacy") + assert await corpus.revision(thread) + assert [message.text for message in (await corpus.messages(thread, before_seq=None)).items] == ["legacy needle"] + assert (await corpus.around("legacy", "old"))["target_seq"] > 0 diff --git a/backend/uv.lock b/backend/uv.lock index c448d25764..cbca50b78f 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -967,7 +967,7 @@ extensions = [] [[package]] name = "deerflow-extension-api" -version = "0.2.11" +version = "0.2.12" source = { editable = "packages/extension-api" } [[package]] diff --git a/config.example.yaml b/config.example.yaml index 0ecd5ec49b..e2a1060bff 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -3724,6 +3724,15 @@ authorization: # See examples/deerflow-extension-example for a package that exercises all five. # # plugins: +# - name: workspace-search +# package: deerflow-extension-workspace-search +# use: deerflow_workspace_search:install +# table_prefix: extension_workspace_search_ +# enabled: true +# # Install examples/deerflow-extension-workspace-search first; restart Gateway. +# # SQLite/Postgres reuse the application database. Memory deployments use: +# config: +# path: .deer-flow/workspace-search.db # - name: example # PEP 621 entry-point name # package: deerflow-extension-example # Python distribution managed by uv # use: deerflow_extension_example:install diff --git a/docs/guide/workspace.md b/docs/guide/workspace.md index abaa5bf448..7791604d4d 100644 --- a/docs/guide/workspace.md +++ b/docs/guide/workspace.md @@ -30,6 +30,17 @@ Dismissing the drawer returns keyboard focus to the control that opened it. The Agents gallery header wraps to the available width so its import and create actions remain visible beside the compact sidebar on narrow screens. +## Search conversations and messages + +Use **Search** in the sidebar or **⌘K / Ctrl+K** to find conversation titles and +visible user/assistant message text, including code. Filter by conversations or +messages and include archived chats. Results show matching excerpts; selecting a +message opens its historical context, with earlier/later navigation and a **Return +to latest messages** button. Type `>` for new chat, settings and keyboard commands. +Title search is built in. Enable message search with the packaged +[workspace search extension](../../examples/deerflow-extension-workspace-search/README.md). +Its first search backfills existing history and shows progress while indexing. + ## Chat Archive Deleting a chat from the sidebar requires confirmation showing its title. Deletion removes the conversation and its files and cannot be undone. diff --git a/examples/deerflow-extension-example/tests/test_plugin.py b/examples/deerflow-extension-example/tests/test_plugin.py index d13935863f..f8ac532187 100644 --- a/examples/deerflow-extension-example/tests/test_plugin.py +++ b/examples/deerflow-extension-example/tests/test_plugin.py @@ -29,6 +29,7 @@ class FakeRegistry: def __init__(self) -> None: self.plugins: list[Any] = [] self.knowledge_providers: list[tuple[str, Any]] = [] + self.workspace_search_providers: list[Any] = [] self.middleware_contributors: list[Any] = [] self.task_lifecycle_contributors: list[Any] = [] self.system_model_observers: list[Any] = [] @@ -45,6 +46,10 @@ class FakeRegistry: self.knowledge_providers.append((name, provider)) return True + def workspace_search(self, provider: Any) -> bool: + self.workspace_search_providers.append(provider) + return True + def middlewares(self, contributor: Any) -> None: self.middleware_contributors.append(contributor) @@ -92,6 +97,7 @@ def test_install_registers_the_five_demonstrated_contribution_kinds() -> None: assert registry.agent_assembly_observers == [] assert registry.context_compaction_observers == [] assert registry.knowledge_providers == [] + assert registry.workspace_search_providers == [] assert registry.plugins == [] assert [route.path for route in registry.contributed_routers[0].routes] == ["/api/extension-example/stats"] assert install.__deerflow_api__ == "0.2.0" diff --git a/examples/deerflow-extension-workspace-search/README.md b/examples/deerflow-extension-workspace-search/README.md new file mode 100644 index 0000000000..828543ee40 --- /dev/null +++ b/examples/deerflow-extension-workspace-search/README.md @@ -0,0 +1,67 @@ +# Workspace message search + +Adds message-body search to the workspace Search / ⌘K / Ctrl+K panel. The host +provides title search, permission checks, highlighted excerpts and deep links; +this package maintains the rebuildable message index. Requires a host with the +extension-api 0.2.12 workspace search contribution. + +From the repository root, install through the existing extension manager: + +```sh +make extension-install SOURCE="$(pwd)/examples/deerflow-extension-workspace-search" +``` + +Add `table_prefix: extension_workspace_search_` to the resulting `plugins:` entry +in `config.yaml`, then restart Gateway. A manually managed installation uses: + +```yaml +plugins: + - name: workspace-search + package: deerflow-extension-workspace-search + use: deerflow_workspace_search:install + enabled: true + table_prefix: extension_workspace_search_ + config: + path: .deer-flow/workspace-search.db +``` + +The application database is reused when configured. `path` is only the fallback +SQLite index location for an in-memory application database. Index tables contain +message text and follow the same storage access controls as conversation data. + +Search uses case-insensitive literal substrings, including Chinese, filenames +inside messages and code identifiers. SQLite uses FTS5 trigram acceleration when +available; one/two-character queries use a literal scan. PostgreSQL uses escaped +literal matching with a GIN trigram index when `pg_trgm` is already installed. +The extension never installs database extensions. Artifact names outside message +text, tool output, hidden controls, internal reasoning and subagent logs are excluded. + +The first nonempty message search starts an owner-scoped historical backfill. +Requests wait at most one second and report indexing progress until ready. Later +searches compare thread revisions and rebuild changed threads; deletion, editing +and regeneration are reconciled before matching. The host rereads every candidate +from visible history before showing its excerpt. The index is not an authorization +source. Both default and custom-agent conversations preserve their own routes. + +The current refresh reads revision metadata for all of the searching user's +threads. Very large histories may spend longer in the indexing state; no external +search service or background worker is required. Searches have cursor pagination +ordered by title match, thread update time and message sequence. + +Validation: `backend/tests/test_workspace_search.py` covers memory, JSONL and +SQLite histories, Unicode/symbol queries, ownership, filters, index refresh and +historical windows. `frontend/tests/e2e-real-backend/workspace-search.spec.ts` +drives the real Gateway and frontend with synthetic history. Run the replay +Gateway with `--workspace-search` to enable this provider in that isolated stack. + +Review follow-up behavior: message jumps use a single owned thread lookup, and +latest message versions win across storage pages. Legacy checkpoint-only +threads are seeded through the host history-migration reservation before +indexing. SQL membership checks use batches of 250 thread IDs. Clean SQLite +FTS projections survive restart; literal-fallback writes mark the projection +dirty and the next FTS-capable startup repairs it. + +Title matching keeps Python Unicode casefold semantics across SQL dialects, +including German sharp-s. It scans metadata in batches of 200 and stops once +the requested result page is filled; a no-match query can still scan the +owner's title metadata. diff --git a/examples/deerflow-extension-workspace-search/deerflow_workspace_search/__init__.py b/examples/deerflow-extension-workspace-search/deerflow_workspace_search/__init__.py new file mode 100644 index 0000000000..c50a6f70de --- /dev/null +++ b/examples/deerflow-extension-workspace-search/deerflow_workspace_search/__init__.py @@ -0,0 +1,10 @@ +"""Install a persistent workspace search index behind the host search contract.""" + +from .index import WorkspaceSearchIndex + + +def install(registry, config): + index = WorkspaceSearchIndex(path=str(config.get("path", ".deer-flow/workspace-search.db"))) + if not registry.workspace_search(index): + raise RuntimeError("This host does not support workspace search providers") + registry.service(index) diff --git a/examples/deerflow-extension-workspace-search/deerflow_workspace_search/index.py b/examples/deerflow-extension-workspace-search/deerflow_workspace_search/index.py new file mode 100644 index 0000000000..79b1d1ba6e --- /dev/null +++ b/examples/deerflow-extension-workspace-search/deerflow_workspace_search/index.py @@ -0,0 +1,227 @@ +"""Rebuildable SQL projection of the host's visible transcript. + +No checkpoint, filesystem conversation scan or streaming-token hook is needed. +Refresh jobs read through an explicitly owner-bound host capability. A request +waits briefly for its refresh, returning indexing=True while historical data is +backfilled. Partial refreshes are never reported as a complete empty result. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +from dataclasses import asdict +from pathlib import Path + +from deerflow_extension_api.search import SearchCorpus, SearchMessage, SearchPage, SearchThread +from sqlalchemy import Column, Integer, MetaData, String, Table, Text, and_, delete, insert, or_, select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +metadata = MetaData() +documents = Table( + "extension_workspace_search_documents", + metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("owner", String(64), nullable=False, index=True), + Column("thread_id", String(64), nullable=False, index=True), + Column("message_id", Text, nullable=False), + Column("seq", Integer, nullable=False), + Column("updated_at", String(64), nullable=False), + Column("normalized", Text, nullable=False), + Column("payload", Text, nullable=False), +) +revisions = Table( + "extension_workspace_search_revisions", + metadata, + Column("owner", String(64), primary_key=True), + Column("thread_id", String(64), primary_key=True), + Column("revision", String(64), nullable=False), +) +fts_state = Table( + "extension_workspace_search_fts_state", + metadata, + Column("key", String(32), primary_key=True), + Column("value", String(16), nullable=False), +) +_FTS = "extension_workspace_search_fts" + + +class WorkspaceSearchIndex: + def __init__(self, *, path: str = ".deer-flow/workspace-search.db", wait_seconds: float = 1.0): + self.path = path + self.wait_seconds = wait_seconds + self._tasks: dict[str, asyncio.Task] = {} + self._engine = None + self._owns_engine = False + self._sf = None + self._fts = False + + async def start(self, deps): + self._sf = deps.session_factory + if self._sf is None: + await asyncio.to_thread(Path(self.path).absolute().parent.mkdir, parents=True, exist_ok=True) + self._engine = create_async_engine(f"sqlite+aiosqlite:///{self.path}") + self._sf = async_sessionmaker(self._engine, expire_on_commit=False) + self._owns_engine = True + else: + self._engine = self._sf.kw["bind"] + async with self._engine.begin() as connection: + await connection.run_sync(metadata.create_all) + if self._engine.dialect.name == "sqlite": + # Older SQLite distributions may lack FTS5/trigram; literal matching + # remains correct, including one/two-character Chinese queries. + from sqlalchemy.exc import OperationalError + + try: + async with self._engine.begin() as connection: + existed = await connection.scalar(text("SELECT name FROM sqlite_master WHERE type='table' AND name=:name"), {"name": _FTS}) + state = await connection.scalar(select(fts_state.c.value).where(fts_state.c.key == "projection")) + await connection.execute(text(f"CREATE VIRTUAL TABLE IF NOT EXISTS {_FTS} USING fts5(docid UNINDEXED, normalized, tokenize='trigram')")) + if not existed or state != "clean": + await connection.execute(text(f"DELETE FROM {_FTS}")) + await connection.execute(text(f"INSERT INTO {_FTS}(docid, normalized) SELECT id, normalized FROM {documents.name}")) + await connection.execute(text(f"INSERT INTO {fts_state.name} (key, value) VALUES ('projection', 'clean') ON CONFLICT(key) DO UPDATE SET value='clean'")) + self._fts = True + except OperationalError as exc: + if "no such module" not in str(exc) and "no such tokenizer" not in str(exc): + raise + elif self._engine.dialect.name == "postgresql": + async with self._engine.begin() as connection: + available = await connection.scalar(text("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')")) + if available: + await connection.execute(text(f"CREATE INDEX IF NOT EXISTS ix_workspace_search_trgm ON {documents.name} USING gin (normalized gin_trgm_ops)")) + + async def stop(self): + tasks = list(self._tasks.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + self._tasks.clear() + if self._owns_engine and self._engine is not None: + await self._engine.dispose() + self._sf = None + + async def _replace(self, owner, thread, revision, messages): + async with self._sf() as session, session.begin(): + if self._engine.dialect.name == "sqlite" and not self._fts: + await session.execute(text(f"INSERT INTO {fts_state.name} (key, value) VALUES ('projection', 'dirty') ON CONFLICT(key) DO UPDATE SET value='dirty'")) + where = and_(documents.c.owner == owner, documents.c.thread_id == thread.thread_id) + if self._fts: + await session.execute(text(f"DELETE FROM {_FTS} WHERE docid IN (SELECT id FROM {documents.name} WHERE owner=:owner AND thread_id=:thread_id)"), {"owner": owner, "thread_id": thread.thread_id}) + await session.execute(delete(documents).where(where)) + if messages: + await session.execute( + insert(documents), + [dict(owner=owner, thread_id=thread.thread_id, message_id=m.message_id, seq=m.seq, updated_at=thread.updated_at, normalized=m.text.casefold(), payload=json.dumps(asdict(m), ensure_ascii=False)) for m in messages], + ) + if self._fts: + await session.execute(text(f"INSERT INTO {_FTS}(docid, normalized) SELECT id, normalized FROM {documents.name} WHERE owner=:owner AND thread_id=:thread_id"), {"owner": owner, "thread_id": thread.thread_id}) + from sqlalchemy.dialects.postgresql import insert as pg_insert + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + + upsert = pg_insert if self._engine.dialect.name == "postgresql" else sqlite_insert + await session.execute(upsert(revisions).values(owner=owner, thread_id=thread.thread_id, revision=revision).on_conflict_do_update(index_elements=["owner", "thread_id"], set_={"revision": revision})) + + async def _refresh(self, corpus: SearchCorpus): + threads = await corpus.threads() + async with self._sf() as session: + known = dict((await session.execute(select(revisions.c.thread_id, revisions.c.revision).where(revisions.c.owner == corpus.user_id))).all()) + for thread in threads: + revision = await corpus.revision(thread) + if revision is None or known.get(thread.thread_id) == revision: + continue + messages: dict[str, SearchMessage] = {} + before = None + while True: + page = await corpus.messages(thread, before_seq=before) + # Backward pages: keep the latest visible version of each ID. + for message in reversed(page.items): + messages.setdefault(message.message_id, message) + if page.before_seq is None: + break + if before is not None and page.before_seq >= before: + raise RuntimeError("Search backfill cursor did not advance") + before = page.before_seq + if await corpus.revision(thread) != revision: + # A concurrent run changed this thread; retry on the next request. + return False + await self._replace(corpus.user_id, thread, revision, messages.values()) + live = {thread.thread_id for thread in threads} + for removed in set(known) - live: + async with self._sf() as session, session.begin(): + if self._engine.dialect.name == "sqlite" and not self._fts: + await session.execute(text(f"INSERT INTO {fts_state.name} (key, value) VALUES ('projection', 'dirty') ON CONFLICT(key) DO UPDATE SET value='dirty'")) + if self._fts: + await session.execute(text(f"DELETE FROM {_FTS} WHERE docid IN (SELECT id FROM {documents.name} WHERE owner=:owner AND thread_id=:thread_id)"), {"owner": corpus.user_id, "thread_id": removed}) + await session.execute(delete(documents).where(documents.c.owner == corpus.user_id, documents.c.thread_id == removed)) + await session.execute(delete(revisions).where(revisions.c.owner == corpus.user_id, revisions.c.thread_id == removed)) + return True + + async def search(self, corpus: SearchCorpus, *, query: str, threads: tuple[SearchThread, ...], limit: int, cursor: str | None = None) -> SearchPage: + if self._sf is None: + raise RuntimeError("Workspace search index is unavailable") + task = self._tasks.get(corpus.user_id) + if task is None: + # Retain completed backfills for the next poll, but bound abandoned + # users' task handles and consume background exceptions. + for owner, previous in list(self._tasks.items()): + if len(self._tasks) < 64: + break + if previous.done(): + del self._tasks[owner] + task = asyncio.create_task(self._refresh(corpus)) + task.add_done_callback(lambda completed: None if completed.cancelled() else completed.exception()) + self._tasks[corpus.user_id] = task + try: + complete = await asyncio.wait_for(asyncio.shield(task), timeout=self.wait_seconds) + except TimeoutError: + return SearchPage(indexing=True) + finally: + if task.done() and self._tasks.get(corpus.user_id) is task: + del self._tasks[corpus.user_id] + if not complete: + return SearchPage(indexing=True) + if not threads: + return SearchPage() + normalized = query.strip().casefold() + # The host binds cursors to the filter set. Live thread membership may + # change between pages without invalidating an otherwise valid cursor. + scope = hashlib.sha256(json.dumps([corpus.user_id, normalized]).encode()).hexdigest() + statement = select(documents).where(documents.c.owner == corpus.user_id, documents.c.normalized.contains(normalized, autoescape=True)) + if self._fts and len(normalized) >= 3: + statement = statement.where(documents.c.id.in_(text(f"SELECT docid FROM {_FTS} WHERE {_FTS} MATCH :match"))).params(match='"' + normalized.replace('"', '""') + '"') + if cursor: + try: + saved_scope, updated, thread_id, seq, message_id = json.loads(base64.urlsafe_b64decode(cursor)) + if saved_scope != scope or not isinstance(seq, int): + raise ValueError + except (ValueError, TypeError, json.JSONDecodeError) as exc: + raise ValueError("Invalid search cursor") from exc + statement = statement.where( + or_( + documents.c.updated_at < updated, + and_(documents.c.updated_at == updated, documents.c.thread_id < thread_id), + and_(documents.c.updated_at == updated, documents.c.thread_id == thread_id, documents.c.seq < seq), + and_(documents.c.updated_at == updated, documents.c.thread_id == thread_id, documents.c.seq == seq, documents.c.message_id < message_id), + ) + ) + statement = statement.order_by(documents.c.updated_at.desc(), documents.c.thread_id.desc(), documents.c.seq.desc(), documents.c.message_id.desc()).limit(limit + 1) + async with self._sf() as session: + rows = [] + ids = [thread.thread_id for thread in threads] + # Keep SQL parameter counts below SQLite's portable limit. Each + # batch contributes only its top page; merge by the same total order. + for offset in range(0, len(ids), 250): + batch = statement.where(documents.c.thread_id.in_(ids[offset : offset + 250])) + rows.extend((await session.execute(batch)).mappings().all()) + rows.sort(key=lambda row: (row["updated_at"], row["thread_id"], row["seq"], row["message_id"]), reverse=True) + rows = rows[: limit + 1] + more = len(rows) > limit + rows = rows[:limit] + next_cursor = None + if more: + last = rows[-1] + next_cursor = base64.urlsafe_b64encode(json.dumps([scope, last["updated_at"], last["thread_id"], last["seq"], last["message_id"]]).encode()).decode() + return SearchPage(tuple(SearchMessage(**json.loads(row["payload"])) for row in rows), next_cursor) diff --git a/examples/deerflow-extension-workspace-search/pyproject.toml b/examples/deerflow-extension-workspace-search/pyproject.toml new file mode 100644 index 0000000000..34eff450d7 --- /dev/null +++ b/examples/deerflow-extension-workspace-search/pyproject.toml @@ -0,0 +1,16 @@ +[project] +name = "deerflow-extension-workspace-search" +version = "0.1.0" +description = "Visible conversation full-text search for DeerFlow" +requires-python = ">=3.12" +dependencies = ["deerflow-extension-api>=0.2.12", "sqlalchemy>=2", "aiosqlite>=0.20"] + +[project.entry-points."deerflow.extensions"] +workspace-search = "deerflow_workspace_search:install" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["deerflow_workspace_search"] diff --git a/frontend/playwright.real-backend.config.ts b/frontend/playwright.real-backend.config.ts index 204d9a7438..3a064ce684 100644 --- a/frontend/playwright.real-backend.config.ts +++ b/frontend/playwright.real-backend.config.ts @@ -35,7 +35,7 @@ export default defineConfig({ webServer: [ { - command: `uv run python scripts/run_replay_gateway.py --port ${gatewayPort} --cors ${frontendUrl}`, + command: `uv run python scripts/run_replay_gateway.py --port ${gatewayPort} --cors ${frontendUrl} --workspace-search`, cwd: "../backend", url: `${gatewayUrl}/health`, reuseExistingServer: !process.env.CI, diff --git a/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx b/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx index 86e1405da7..10f6992561 100644 --- a/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx +++ b/frontend/src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx @@ -23,12 +23,10 @@ import { type InputBoxSubmitOptions, } from "@/components/workspace/input-box"; import { KnowledgeScopeSelector } from "@/components/workspace/knowledge-scope-selector"; -import { - MessageList, - MESSAGE_LIST_DEFAULT_PADDING_BOTTOM, -} from "@/components/workspace/messages"; +import { MESSAGE_LIST_DEFAULT_PADDING_BOTTOM } from "@/components/workspace/messages"; import { ThreadContext } from "@/components/workspace/messages/context"; import { HumanInputCard } from "@/components/workspace/messages/human-input-card"; +import { SearchableMessageList as MessageList } from "@/components/workspace/search-message-window"; import { SidecarProvider, SidecarTrigger, @@ -371,6 +369,17 @@ export default function AgentChatPage() { } const executionOptions = { ...scopedOptions, + onSent: () => { + const url = new URL(window.location.href); + url.searchParams.delete("message"); + url.searchParams.delete("message_seq"); + history.replaceState( + null, + "", + `${url.pathname}${url.search}${url.hash}`, + ); + scopedOptions?.onSent?.(); + }, additionalKwargs: { ...scopedOptions?.additionalKwargs, ...modelInput.additionalKwargs, diff --git a/frontend/src/components/workspace/AGENTS.md b/frontend/src/components/workspace/AGENTS.md index 4548e84a58..808ecc3622 100644 --- a/frontend/src/components/workspace/AGENTS.md +++ b/frontend/src/components/workspace/AGENTS.md @@ -78,3 +78,8 @@ pnpm catalog:sync # after a change to the capability catalog a large file. - **The parser is told the record separator explicitly.** Auto-detection is corrupted by a newline inside an incomplete quoted field of a truncated sample. + +## Conversation search + +Keep historical message windows separate from live stream history. After +the composer accepts a new turn, return to the live conversation. diff --git a/frontend/src/components/workspace/chats/chat-page.tsx b/frontend/src/components/workspace/chats/chat-page.tsx index 29e15320e5..822a9c118b 100644 --- a/frontend/src/components/workspace/chats/chat-page.tsx +++ b/frontend/src/components/workspace/chats/chat-page.tsx @@ -18,12 +18,10 @@ import { type InputBoxSubmitOptions, } from "@/components/workspace/input-box"; import { KnowledgeScopeSelector } from "@/components/workspace/knowledge-scope-selector"; -import { - MessageList, - MESSAGE_LIST_DEFAULT_PADDING_BOTTOM, -} from "@/components/workspace/messages"; +import { MESSAGE_LIST_DEFAULT_PADDING_BOTTOM } from "@/components/workspace/messages"; import { ThreadContext } from "@/components/workspace/messages/context"; import { HumanInputCard } from "@/components/workspace/messages/human-input-card"; +import { SearchableMessageList as MessageList } from "@/components/workspace/search-message-window"; import { SidecarProvider, SidecarTrigger, @@ -449,6 +447,17 @@ export default function ChatPage() { } const executionOptions = { ...scopedOptions, + onSent: () => { + const url = new URL(window.location.href); + url.searchParams.delete("message"); + url.searchParams.delete("message_seq"); + history.replaceState( + null, + "", + `${url.pathname}${url.search}${url.hash}`, + ); + scopedOptions?.onSent?.(); + }, additionalKwargs: { ...scopedOptions?.additionalKwargs, ...modelInput.additionalKwargs, diff --git a/frontend/src/components/workspace/command-palette.tsx b/frontend/src/components/workspace/command-palette.tsx index 38bbdc3050..04a4291ee4 100644 --- a/frontend/src/components/workspace/command-palette.tsx +++ b/frontend/src/components/workspace/command-palette.tsx @@ -1,6 +1,10 @@ "use client"; import { + ArchiveIcon, + Loader2Icon, + SearchIcon, + MessagesSquareIcon, KeyboardIcon, MessageSquarePlusIcon, SettingsIcon, @@ -8,9 +12,9 @@ import { import { useRouter } from "next/navigation"; import { useCallback, useEffect, useMemo, useState } from "react"; +import { Button } from "@/components/ui/button"; import { - CommandDialog, - CommandEmpty, + Command, CommandGroup, CommandInput, CommandItem, @@ -25,6 +29,12 @@ import { DialogTitle, } from "@/components/ui/dialog"; import { useI18n } from "@/core/i18n/hooks"; +import { + OPEN_SEARCH_EVENT, + pathOfSearchHit, + type SearchFilters, +} from "@/core/search/api"; +import { useWorkspaceSearch } from "@/core/search/hooks"; import { useGlobalShortcuts } from "@/hooks/use-global-shortcuts"; import { useSettingsDialog } from "./settings"; @@ -34,6 +44,24 @@ export function CommandPalette() { const router = useRouter(); const { openSettings } = useSettingsDialog(); const [open, setOpen] = useState(false); + const [input, setInput] = useState(""); + const [composing, setComposing] = useState(false); + const [filters, setFilters] = useState({ + type: "all", + includeArchived: true, + }); + const commandMode = input.trimStart().startsWith(">"); + const search = useWorkspaceSearch( + input, + filters, + open && !commandMode && !composing, + ); + const busy = search.waiting || search.isPending || search.isFetching; + useEffect(() => { + const show = () => setOpen(true); + window.addEventListener(OPEN_SEARCH_EVENT, show); + return () => window.removeEventListener(OPEN_SEARCH_EVENT, show); + }, []); const [shortcutsOpen, setShortcutsOpen] = useState(false); const [isMac, setIsMac] = useState(false); @@ -72,32 +100,300 @@ export function CommandPalette() { return ( <> - - - - {t.shortcuts.noResults} - - - - {t.sidebar.newChat} - - {metaKey} - {shiftKey}N - - - - - {t.common.settings} - {metaKey}, - - - - {t.shortcuts.keyboardShortcuts} - {metaKey}/ - - - - + + + + {t.workspaceSearch.title} + + + {t.workspaceSearch.description} + + { + if ( + event.key === "Enter" && + (composing || + event.nativeEvent.isComposing || + event.nativeEvent.keyCode === 229) + ) { + event.preventDefault(); + event.stopPropagation(); + } + }} + > + setComposing(true)} + onCompositionEnd={() => setComposing(false)} + /> + {!commandMode && ( +
+ {(["all", "thread", "message"] as const).map((type) => ( + + ))} + +
+ )} + + {!commandMode && ( + <> + {!search.supported && ( +

+ {t.workspaceSearch.unsupported} +

+ )} + {search.supported && search.isError && ( +
+

{t.workspaceSearch.error}

+ +
+ )} + {search.supported && + !search.isError && + busy && + search.items.length === 0 && ( +
+ + {t.workspaceSearch.searching} +
+ )} + {search.data?.pages.some((page) => page.indexing) && ( +

+ {t.workspaceSearch.indexing} +

+ )} + {input.trim() && + search.data?.pages[0]?.messages_available === false && ( +

+ {t.workspaceSearch.unavailable} +

+ )} + {(["thread", "message"] as const).map((type) => { + const hits = search.items.filter( + (hit) => hit.type === type, + ); + if (!hits.length) return null; + return ( + + {hits.map((hit) => ( + { + router.push(pathOfSearchHit(hit)); + setOpen(false); + }} + > + {type === "thread" ? ( + + ) : ( + + )} +
+
+ + {hit.title.trim() + ? hit.title + : t.workspaceSearch.untitled} + + {hit.archived && ( + + + {t.workspaceSearch.archived} + + )} +
+ {type === "message" && ( +

+ {hit.snippet.map((segment, index) => + segment.matched ? ( + + {segment.text} + + ) : ( + {segment.text} + ), + )} +

+ )} +
+ {type === "message" && ( + + {hit.role === "user" + ? t.workspaceSearch.user + : t.workspaceSearch.assistant} + + )} + + {hit.updated_at + ? new Date( + hit.updated_at, + ).toLocaleDateString() + : ""} + +
+
+ + ))} + + ); + })} + {!busy && + !search.isError && + search.supported && + search.items.length === 0 && + !search.data?.pages.some((page) => page.indexing) && ( +

+ {t.workspaceSearch.empty} +

+ )} + {search.hasNextPage && ( + void search.fetchNextPage()} + > + {t.workspaceSearch.more} + + )} + + )} + {commandMode && + ![ + t.sidebar.newChat, + t.common.settings, + t.shortcuts.keyboardShortcuts, + ].some((label) => + label + .toLocaleLowerCase() + .includes( + input.trimStart().slice(1).trim().toLocaleLowerCase(), + ), + ) && ( +
+ {t.workspaceSearch.empty} +
+ )} + {(commandMode || !input.trim()) && ( + + {[ + { + label: t.sidebar.newChat, + icon: MessageSquarePlusIcon, + action: handleNewChat, + shortcut: `${metaKey}${shiftKey}N`, + }, + { + label: t.common.settings, + icon: SettingsIcon, + action: handleOpenSettings, + shortcut: `${metaKey},`, + }, + { + label: t.shortcuts.keyboardShortcuts, + icon: KeyboardIcon, + action: handleShowShortcuts, + shortcut: `${metaKey}/`, + }, + ] + .filter( + (action) => + !commandMode || + action.label + .toLocaleLowerCase() + .includes( + input + .trimStart() + .slice(1) + .trim() + .toLocaleLowerCase(), + ), + ) + .map((action) => ( + + + {action.label} + {action.shortcut} + + ))} + + )} + +
+ {t.workspaceSearch.hint} +
+ + + diff --git a/frontend/src/components/workspace/search-message-window.tsx b/frontend/src/components/workspace/search-message-window.tsx new file mode 100644 index 0000000000..8040888c6b --- /dev/null +++ b/frontend/src/components/workspace/search-message-window.tsx @@ -0,0 +1,202 @@ +"use client"; + +import { useQuery } from "@tanstack/react-query"; +import { ArrowDownIcon, ArrowUpIcon, Loader2Icon } from "lucide-react"; +import { usePathname, useRouter, useSearchParams } from "next/navigation"; +import { useEffect, useRef, useState, type ComponentProps } from "react"; +import type { ImgHTMLAttributes } from "react"; + +import { Button } from "@/components/ui/button"; +import { resolveMessageImageURL } from "@/core/artifacts/utils"; +import { useAuth } from "@/core/auth/AuthProvider"; +import { useI18n } from "@/core/i18n/hooks"; +import { fetchMessageWindow } from "@/core/search/api"; +import { cn } from "@/lib/utils"; + +import { MarkdownContent } from "./messages/markdown-content"; +import { createMarkdownLinkComponent } from "./messages/markdown-link"; +import { MessageList } from "./messages/message-list"; + +export function SearchableMessageList( + props: ComponentProps, +) { + const params = useSearchParams(); + const messageId = params.get("message"); + const seq = Number(params.get("message_seq")); + if (!messageId) return ; + return ( + 0 ? seq : undefined} + /> + ); +} + +function SearchMessageWindow({ + threadId, + messageId, + seq, +}: { + threadId: string; + messageId: string; + seq?: number; +}) { + const { t } = useI18n(); + const { user } = useAuth(); + const router = useRouter(); + const pathname = usePathname(); + const params = useSearchParams(); + const target = useRef(null); + const viewport = useRef(null); + const [anchor, setAnchor] = useState({ messageId, seq }); + const query = useQuery({ + queryKey: [ + "search-message-window", + user?.id, + threadId, + anchor.messageId, + anchor.seq, + ], + queryFn: ({ signal }) => + fetchMessageWindow(threadId, anchor.messageId, anchor.seq, signal), + retry: false, + refetchOnWindowFocus: false, + staleTime: 0, + }); + useEffect(() => { + if (query.data && target.current && viewport.current) { + viewport.current.scrollTop = + target.current.offsetTop - + (viewport.current.clientHeight - target.current.offsetHeight) / 2; + } + }, [query.data]); + const returnToLatest = () => { + const next = new URLSearchParams(params.toString()); + next.delete("message"); + next.delete("message_seq"); + const queryString = next.toString(); + router.replace(`${pathname}${queryString ? `?${queryString}` : ""}`, { + scroll: false, + }); + }; + const markdownComponents = { + a: createMarkdownLinkComponent(threadId), + img: ({ src, alt, ...props }: ImgHTMLAttributes) => ( + {alt} + ), + }; + const data = query.data; + const first = data?.messages[0]; + const last = data?.messages.at(-1); + return ( +
+
+ {t.workspaceSearch.context} + +
+
+ {query.isPending && ( +
+ + {t.workspaceSearch.loadingContext} +
+ )} + {query.isError && ( +
+

{t.workspaceSearch.messageUnavailable}

+ +
+ )} + {data && ( +
+ {data.has_before && first && ( + + )} + {data.messages.map((message) => ( +
+
+ + {message.role === "user" + ? t.workspaceSearch.user + : t.workspaceSearch.assistant} + + {message.seq === data.target_seq && ( + {t.workspaceSearch.matchedMessage} + )} +
+ {message.role === "assistant" ? ( + + ) : ( +
+ {message.text} +
+ )} +
+ ))} + {data.has_after && last && ( + + )} +
+ )} +
+
+ ); +} diff --git a/frontend/src/components/workspace/workspace-nav-chat-list.tsx b/frontend/src/components/workspace/workspace-nav-chat-list.tsx index faec9de57e..8377e463b1 100644 --- a/frontend/src/components/workspace/workspace-nav-chat-list.tsx +++ b/frontend/src/components/workspace/workspace-nav-chat-list.tsx @@ -1,6 +1,7 @@ "use client"; import { + SearchIcon, BotIcon, CalendarClock, MessagesSquare, @@ -9,12 +10,14 @@ import { } from "lucide-react"; import Link from "next/link"; import { usePathname } from "next/navigation"; +import { useEffect, useState } from "react"; import { SidebarGroup, SidebarMenu, SidebarMenuButton, SidebarMenuItem, + useSidebar, } from "@/components/ui/sidebar"; import { Tooltip, @@ -25,6 +28,7 @@ import { useAgentsApiEnabled } from "@/core/agents"; import { useAuth } from "@/core/auth/AuthProvider"; import { hasPermission, PERMISSIONS } from "@/core/auth/permissions"; import { useI18n } from "@/core/i18n/hooks"; +import { OPEN_SEARCH_EVENT } from "@/core/search/api"; import { isStaticWebsiteOnly } from "@/core/static-mode"; export function WorkspaceNavChatList() { @@ -32,6 +36,11 @@ export function WorkspaceNavChatList() { const pathname = usePathname(); const { enabled: agentsEnabled } = useAgentsApiEnabled(); const { user } = useAuth(); + const { setOpenMobile } = useSidebar(); + const [shortcut, setShortcut] = useState("⌘K"); + useEffect(() => { + setShortcut(navigator.userAgent.includes("Mac") ? "⌘K" : "Ctrl K"); + }, []); const usageEnabled = !!user && hasPermission(user, PERMISSIONS.RUNS_READ) && @@ -39,6 +48,23 @@ export function WorkspaceNavChatList() { return ( + + { + setOpenMobile(false); + window.dispatchEvent(new Event(OPEN_SEARCH_EVENT)); + }} + aria-label={t.workspaceSearch.open} + tooltip={t.workspaceSearch.title} + className="text-muted-foreground" + > + + {t.workspaceSearch.open} + + {shortcut} + + + diff --git a/frontend/src/core/extensions/catalog.ts b/frontend/src/core/extensions/catalog.ts index ae63d24372..0c1896a14c 100644 --- a/frontend/src/core/extensions/catalog.ts +++ b/frontend/src/core/extensions/catalog.ts @@ -11,6 +11,11 @@ export const extensionCatalog = [ namespace: "office-preview", icon: "file-text", }, + { + key: "workspaceSearch", + package: "deerflow-extension-workspace-search", + namespace: null, + }, { key: "agentTeams", icon: "users", diff --git a/frontend/src/core/i18n/locales/en-US.ts b/frontend/src/core/i18n/locales/en-US.ts index 11465fc802..9afa71d0c4 100644 --- a/frontend/src/core/i18n/locales/en-US.ts +++ b/frontend/src/core/i18n/locales/en-US.ts @@ -124,6 +124,11 @@ export const enUS: Translations = { }, catalog: { officePreview: bundledCopy("deerflow-extension-office-preview", "en-US"), + workspaceSearch: { + title: "Conversation search", + description: + "Search historical message text and jump to the matching conversation turn.", + }, agentTeams: bundledCopy("deerflow-extension-agent-teams", "en-US"), bookmarks: bundledCopy("deerflow-extension-bookmarks", "en-US"), context: bundledCopy("deerflow-extension-jev-context", "en-US"), @@ -2055,6 +2060,39 @@ export const enUS: Translations = { }, // Shortcuts + workspaceSearch: { + title: "Search conversations", + description: "Find conversations and messages, or run a command.", + placeholder: "Search titles and messages…", + open: "Search", + all: "All", + threads: "Conversations", + messages: "Messages", + includeArchived: "Include archived", + archived: "Archived", + recent: "Recent conversations", + searching: "Searching…", + indexing: "Preparing message search… Results will appear here shortly.", + unavailable: + "Message search is unavailable on this installation. You can search conversation titles.", + unsupported: "Search is unavailable in this view.", + error: "Search could not finish. Try again.", + empty: "No matches. Try another keyword or include archived conversations.", + more: "Show more results", + retry: "Try again", + hint: "↑ ↓ to navigate · Enter to open · > for commands", + user: "You", + assistant: "Assistant", + context: "Message context", + returnToLatest: "Return to latest messages", + loadingContext: "Loading message context…", + messageUnavailable: + "This message is no longer available. Return to the conversation to view its current history.", + matchedMessage: "Located message", + earlier: "Earlier messages", + later: "Later messages", + untitled: "Untitled conversation", + }, shortcuts: { searchActions: "Search actions...", noResults: "No results found.", diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts index aeab4303d0..a5c503f9f6 100644 --- a/frontend/src/core/i18n/locales/types.ts +++ b/frontend/src/core/i18n/locales/types.ts @@ -99,6 +99,7 @@ export interface Translations { }; catalog: Record< | "officePreview" + | "workspaceSearch" | "workflows" | "reportSections" | "agentTeams" @@ -1749,6 +1750,37 @@ export interface Translations { }; // Shortcuts + workspaceSearch: { + title: string; + description: string; + placeholder: string; + open: string; + all: string; + threads: string; + messages: string; + includeArchived: string; + archived: string; + recent: string; + searching: string; + indexing: string; + unavailable: string; + unsupported: string; + error: string; + empty: string; + more: string; + retry: string; + hint: string; + user: string; + assistant: string; + context: string; + returnToLatest: string; + loadingContext: string; + messageUnavailable: string; + matchedMessage: string; + earlier: string; + later: string; + untitled: string; + }; shortcuts: { searchActions: string; noResults: string; diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts index 4909d80efb..5f2816aa9a 100644 --- a/frontend/src/core/i18n/locales/zh-CN.ts +++ b/frontend/src/core/i18n/locales/zh-CN.ts @@ -117,6 +117,10 @@ export const zhCN: Translations = { }, catalog: { officePreview: bundledCopy("deerflow-extension-office-preview", "zh-CN"), + workspaceSearch: { + title: "会话全文搜索", + description: "检索历史消息正文，并跳转到命中的对话轮次。", + }, agentTeams: bundledCopy("deerflow-extension-agent-teams", "zh-CN"), bookmarks: bundledCopy("deerflow-extension-bookmarks", "zh-CN"), context: bundledCopy("deerflow-extension-jev-context", "zh-CN"), @@ -1940,6 +1944,37 @@ export const zhCN: Translations = { }, // Shortcuts + workspaceSearch: { + title: "搜索对话", + description: "查找历史对话和消息，或执行快捷操作。", + placeholder: "搜索标题和消息内容…", + open: "搜索", + all: "全部", + threads: "对话", + messages: "消息", + includeArchived: "包含已归档", + archived: "已归档", + recent: "最近对话", + searching: "正在搜索…", + indexing: "正在准备历史消息搜索，结果将自动显示。", + unavailable: "当前部署尚未启用消息搜索，可以搜索对话标题。", + unsupported: "当前视图暂不支持搜索。", + error: "搜索未能完成，请重试。", + empty: "没有找到匹配内容，试试其他关键词或包含已归档对话。", + more: "显示更多结果", + retry: "重试", + hint: "↑ ↓ 选择 · Enter 打开 · 输入 > 查找操作", + user: "你", + assistant: "助手", + context: "消息上下文", + returnToLatest: "返回最新消息", + loadingContext: "正在加载消息上下文…", + messageUnavailable: "这条消息已不可用，可返回对话查看当前历史。", + matchedMessage: "已定位消息", + earlier: "更早的消息", + later: "后续消息", + untitled: "未命名对话", + }, shortcuts: { searchActions: "搜索操作...", noResults: "未找到结果。", diff --git a/frontend/src/core/search/api.ts b/frontend/src/core/search/api.ts new file mode 100644 index 0000000000..3355b782d6 --- /dev/null +++ b/frontend/src/core/search/api.ts @@ -0,0 +1,91 @@ +import { fetch } from "@/core/api/fetcher"; +import { getBackendBaseURL } from "@/core/config"; +import { pathOfThread } from "@/core/threads/utils"; + +export const OPEN_SEARCH_EVENT = "deerflow:open-search"; + +export type SearchSegment = { text: string; matched?: boolean }; +export type WorkspaceSearchHit = { + type: "thread" | "message"; + thread_id: string; + title: string; + agent_name: string | null; + archived: boolean; + updated_at: string; + project_id: string | null; + message_id?: string; + seq?: number; + role?: "user" | "assistant"; + snippet: SearchSegment[]; +}; +export type WorkspaceSearchPage = { + items: WorkspaceSearchHit[]; + cursor: string | null; + indexing: boolean; + messages_available: boolean; +}; +export type SearchFilters = { + type: "all" | "thread" | "message"; + includeArchived: boolean; +}; + +export async function searchWorkspace( + query: string, + filters: SearchFilters, + cursor: string | null, + signal: AbortSignal, +): Promise { + const response = await fetch(`${getBackendBaseURL()}/api/search`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + query, + types: filters.type === "all" ? ["thread", "message"] : [filters.type], + archived: filters.includeArchived ? null : false, + limit: 20, + cursor, + }), + signal, + }); + if (!response.ok) throw new Error(`Search failed (${response.status})`); + return (await response.json()) as WorkspaceSearchPage; +} + +export function pathOfSearchHit(hit: WorkspaceSearchHit): string { + const path = pathOfThread(hit.thread_id, { + agent_name: hit.agent_name ?? undefined, + }); + if (hit.type === "thread" || !hit.message_id) return path; + const params = new URLSearchParams({ message: hit.message_id }); + if (hit.seq !== undefined) params.set("message_seq", String(hit.seq)); + return `${path}?${params}`; +} + +export type MessageWindow = { + messages: { + message_id: string; + seq: number; + role: "user" | "assistant"; + text: string; + created_at: string; + }[]; + target_seq: number; + has_before: boolean; + has_after: boolean; +}; + +export async function fetchMessageWindow( + threadId: string, + messageId: string, + seq: number | undefined, + signal: AbortSignal, +): Promise { + const params = new URLSearchParams({ message_id: messageId }); + if (seq !== undefined) params.set("seq", String(seq)); + const response = await fetch( + `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/messages/around?${params}`, + { signal }, + ); + if (!response.ok) throw new Error(`Message unavailable (${response.status})`); + return (await response.json()) as MessageWindow; +} diff --git a/frontend/src/core/search/hooks.ts b/frontend/src/core/search/hooks.ts new file mode 100644 index 0000000000..f4800bbaa0 --- /dev/null +++ b/frontend/src/core/search/hooks.ts @@ -0,0 +1,65 @@ +"use client"; + +import { useInfiniteQuery, type InfiniteData } from "@tanstack/react-query"; +import { useEffect, useState } from "react"; + +import { useAuth } from "@/core/auth/AuthProvider"; +import { hasPermission } from "@/core/auth/permissions"; +import { isStaticWebsiteOnly } from "@/core/static-mode"; + +import { + searchWorkspace, + type SearchFilters, + type WorkspaceSearchPage, +} from "./api"; + +export function useWorkspaceSearch( + input: string, + filters: SearchFilters, + enabled: boolean, +) { + const { user } = useAuth(); + const [settled, setSettled] = useState(input); + useEffect(() => { + const timer = setTimeout(() => setSettled(input), 250); + return () => clearTimeout(timer); + }, [input]); + const query = input.trim(); + const ready = settled.trim() === query; + const supported = + !isStaticWebsiteOnly() && hasPermission(user, "threads:read"); + const scope = JSON.stringify([ + user?.id, + supported, + user?.permissions, + filters, + ]); + const result = useInfiniteQuery< + WorkspaceSearchPage, + Error, + InfiniteData, + string[], + string | null + >({ + queryKey: ["workspace-search", scope, query], + enabled: enabled && ready && supported, + placeholderData: (previous, previousQuery) => + previousQuery?.queryKey[1] === scope ? previous : undefined, + initialPageParam: null as string | null, + queryFn: ({ pageParam, signal }) => + searchWorkspace(query, filters, pageParam, signal), + getNextPageParam: (page) => page.cursor ?? undefined, + staleTime: 0, + gcTime: 60_000, + retry: false, + refetchOnWindowFocus: false, + refetchInterval: (state) => + state.state.data?.pages.some((page) => page.indexing) ? 600 : false, + }); + return { + ...result, + supported, + waiting: !ready, + items: result.data?.pages.flatMap((page) => page.items) ?? [], + }; +} diff --git a/frontend/tests/e2e-real-backend/workspace-search.spec.ts b/frontend/tests/e2e-real-backend/workspace-search.spec.ts new file mode 100644 index 0000000000..a7acdf5fa5 --- /dev/null +++ b/frontend/tests/e2e-real-backend/workspace-search.spec.ts @@ -0,0 +1,81 @@ +import { expect, test } from "@playwright/test"; + +test("search titles and old message bodies, navigate both directions and return to live history", async ({ + page, + request, +}, testInfo) => { + const threadId = `search-${Date.now()}`; + const title = "新能源行业调研 · 2026 秋季"; + const needle = "储能成本测算"; + const created = await request.post("/api/threads", { + data: { thread_id: threadId, metadata: { title } }, + }); + expect(created.status(), await created.text()).toBe(200); + const renamed = await request.post(`/api/threads/${threadId}/state`, { + data: { values: { title } }, + }); + expect(renamed.ok(), await renamed.text()).toBeTruthy(); + const messages = Array.from({ length: 140 }, (_, i) => ({ + id: `search-message-${i}`, + role: i % 2 ? "ai" : "human", + content: + i === 5 + ? `## ${needle}\n\n报告比较了不同储能路线的单位成本与回收周期。\n\n\`\`\`python\ndef payback_years(capex, annual_savings):\n return capex / annual_savings\n\`\`\`\n\n建议采用分阶段投资方案，并持续跟踪电价变化。` + : `${i % 2 ? "分析结果" : "后续问题"} ${i}：比较不同技术路线与市场数据。`, + })); + const seeded = await request.post("/api/test-only/seed-runs", { + data: { + thread_id: threadId, + runs: [ + { + run_id: `${threadId}-run`, + created_at: "2026-10-25T08:00:00Z", + messages, + }, + ], + }, + }); + expect(seeded.ok(), await seeded.text()).toBeTruthy(); + await page.goto("/workspace/chats/new"); + await page + .getByRole("button", { name: /^Search/ }) + .first() + .click(); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + const input = dialog.getByPlaceholder("Search titles and messages…"); + await input.fill("新能源"); + await expect( + dialog.getByRole("option").filter({ hasText: title }).first(), + ).toBeVisible(); + await input.fill(needle); + const result = dialog.getByRole("option").filter({ hasText: needle }).first(); + await expect(result).toBeVisible({ timeout: 30_000 }); + await expect(result.locator("mark")).toHaveText(needle); + await page.screenshot({ path: testInfo.outputPath("search-panel.png") }); + await result.click(); + await expect(page).toHaveURL(/message=search-message-5&message_seq=6/); + const window = page.getByTestId("search-message-window"); + const target = window.locator('[data-search-target="true"]'); + await expect(target).toContainText(needle); + await expect(target).toBeInViewport(); + await expect(window.locator("article")).toHaveCount(26); + await page.screenshot({ path: testInfo.outputPath("message-jump.png") }); + await window.getByRole("button", { name: "Later messages" }).click(); + await expect(target).toHaveAttribute("data-message-id", "search-message-25"); + await window.getByRole("button", { name: "Earlier messages" }).click(); + await expect(target).toHaveAttribute("data-message-id", "search-message-5"); + await page.reload(); + await expect(window.locator('[data-search-target="true"]')).toContainText( + needle, + ); + await window + .getByRole("button", { name: "Return to latest messages" }) + .click(); + await expect(page).not.toHaveURL(/message=/); + await expect(window).toBeHidden(); + await expect( + page.getByText("分析结果 139：比较不同技术路线与市场数据。", { + exact: true, + }), + ).toBeVisible(); +}); diff --git a/frontend/tests/e2e/bookmark-plugin.spec.ts b/frontend/tests/e2e/bookmark-plugin.spec.ts index 75b8bbcef8..8a66fe50b2 100644 --- a/frontend/tests/e2e/bookmark-plugin.spec.ts +++ b/frontend/tests/e2e/bookmark-plugin.spec.ts @@ -274,11 +274,12 @@ for (const source of ["default", "custom-toolbar", "custom-sidebar"]) { await page .getByRole("searchbox", { name: "Search bookmarks", exact: true }) .fill("ORCHID"); - await page.getByRole("button", { name: "Search", exact: true }).click(); + const bookmarkSearch = page + .getByRole("region", { name: "Bookmarks", exact: true }) + .getByRole("button", { name: "Search", exact: true }); + await bookmarkSearch.click(); await expect(page.getByText(/1 saved/)).toBeVisible(); - await expect( - page.getByRole("button", { name: "Search", exact: true }), - ).toBeEnabled(); + await expect(bookmarkSearch).toBeEnabled(); await page .getByRole("button", { name: "Open conversation", exact: true }) .click(); diff --git a/frontend/tests/e2e/workspace-search-review.spec.ts b/frontend/tests/e2e/workspace-search-review.spec.ts new file mode 100644 index 0000000000..19c9ad9763 --- /dev/null +++ b/frontend/tests/e2e/workspace-search-review.spec.ts @@ -0,0 +1,191 @@ +import { expect, test } from "@playwright/test"; + +import { mockLangGraphAPI, MOCK_THREAD_ID } from "./utils/mock-api"; + +for (const agentName of [undefined, "researcher"]) { + test(`new submissions leave historical search: ${agentName ?? "default"}`, async ({ + page, + }) => { + test.setTimeout(60000); + mockLangGraphAPI(page, { + agents: [{ name: "researcher" }], + threads: [ + { + thread_id: MOCK_THREAD_ID, + title: "Search chat", + agent_name: agentName, + messages: [{ id: "old", type: "ai", content: "older answer" }], + }, + ], + }); + const pathname = agentName + ? `/workspace/agents/${agentName}/chats/${MOCK_THREAD_ID}` + : `/workspace/chats/${MOCK_THREAD_ID}`; + await page.route("**/api/search", (route) => + route.fulfill({ + json: { + items: [ + { + type: "message", + thread_id: MOCK_THREAD_ID, + title: "Historical match", + agent_name: agentName, + message_id: "old", + seq: 1, + role: "assistant", + snippet: [{ text: "older answer" }], + }, + ], + cursor: null, + indexing: false, + messages_available: true, + }, + }), + ); + await page.route("**/messages/around?**", (route) => + route.fulfill({ + json: { + messages: [ + { + thread_id: MOCK_THREAD_ID, + message_id: "old", + seq: 1, + role: "assistant", + text: "older answer", + }, + ], + target_seq: 1, + has_before: false, + has_after: false, + }, + }), + ); + await page.goto(pathname); + const composer = page.getByRole("textbox", { + name: "How can I assist you today?", + }); + await expect(composer).toBeEditable(); + const first = page.waitForResponse( + (r) => + r.request().method() === "POST" && r.url().includes("/runs/stream"), + ); + await composer.fill("first turn"); + await composer.press("Enter"); + await (await first).finished(); + await expect( + page.getByRole("button", { name: "Submit", exact: true }), + ).toBeEnabled(); + await page.keyboard.press("ControlOrMeta+k"); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + await dialog.getByPlaceholder("Search titles and messages…").fill("older"); + await dialog + .getByRole("option") + .filter({ hasText: "Historical match" }) + .click(); + await expect(page.getByTestId("search-message-window")).toBeVisible(); + const second = page.waitForResponse( + (r) => + r.request().method() === "POST" && r.url().includes("/runs/stream"), + ); + await composer.fill("second turn"); + await composer.press("Enter"); + await (await second).finished(); + await expect(page).not.toHaveURL(/message=old/); + await expect(page.getByTestId("search-message-window")).toHaveCount(0); + await expect(page.getByText("second turn", { exact: true })).toBeVisible(); + }); +} + +test("return to latest preserves unrelated parameters without URLSearchParams.size", async ({ + page, +}) => { + mockLangGraphAPI(page, { + threads: [{ thread_id: MOCK_THREAD_ID, title: "Audit" }], + }); + await page.addInitScript(() => + Object.defineProperty(URLSearchParams.prototype, "size", { + configurable: true, + get: () => undefined, + }), + ); + await page.route("**/messages/around?**", (route) => + route.fulfill({ + json: { + messages: [ + { + thread_id: MOCK_THREAD_ID, + message_id: "old", + seq: 1, + role: "assistant", + text: "Old text", + }, + ], + target_seq: 1, + has_before: false, + has_after: false, + }, + }), + ); + await page.goto( + `/workspace/chats/${MOCK_THREAD_ID}?message=old&message_seq=1&custom=keep`, + ); + await expect(page.getByTestId("search-message-window")).toBeVisible(); + await page.getByRole("button", { name: "Return to latest messages" }).click(); + await expect(page).toHaveURL( + new RegExp(`/workspace/chats/${MOCK_THREAD_ID}\\?custom=keep$`), + ); +}); + +test("historical markdown resolves thread artifact links and images", async ({ + page, +}) => { + mockLangGraphAPI(page, { + threads: [{ thread_id: MOCK_THREAD_ID, title: "Artifacts" }], + }); + await page.route("**/messages/around?**", (route) => + route.fulfill({ + json: { + messages: [ + { + thread_id: MOCK_THREAD_ID, + message_id: "old", + seq: 1, + role: "assistant", + text: "[Report](/mnt/user-data/outputs/report.md)\n\n![Chart](/mnt/user-data/outputs/chart.png)", + }, + ], + target_seq: 1, + has_before: false, + has_after: false, + }, + }), + ); + await page.goto( + `/workspace/chats/${MOCK_THREAD_ID}?message=old&message_seq=1`, + ); + const window = page.getByTestId("search-message-window"); + await expect( + window.getByRole("link", { name: "Report", exact: true }), + ).toHaveAttribute( + "href", + /\/api\/threads\/[^/]+\/artifacts\/mnt\/user-data\/outputs\/report.md$/, + ); + await expect(window.getByRole("img", { name: "Chart" })).toHaveAttribute( + "src", + /\/api\/threads\/[^/]+\/artifacts\/mnt\/user-data\/outputs\/chart.png$/, + ); +}); + +test("command mode reports a query without matching commands", async ({ + page, +}) => { + mockLangGraphAPI(page); + await page.goto("/workspace/chats/new"); + await page.keyboard.press("ControlOrMeta+k"); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + await dialog + .getByPlaceholder("Search titles and messages…") + .fill(">no-such-command"); + await expect(dialog.getByRole("status")).toBeVisible(); + await expect(dialog.getByRole("option")).toHaveCount(0); +}); diff --git a/frontend/tests/e2e/workspace-search.spec.ts b/frontend/tests/e2e/workspace-search.spec.ts new file mode 100644 index 0000000000..3407e83102 --- /dev/null +++ b/frontend/tests/e2e/workspace-search.spec.ts @@ -0,0 +1,227 @@ +import { expect, test } from "@playwright/test"; + +import { mockLangGraphAPI, MOCK_THREAD_ID } from "./utils/mock-api"; + +test.beforeEach(async ({ page }) => { + mockLangGraphAPI(page, { + agents: [{ name: "researcher" }], + threads: [ + { + thread_id: MOCK_THREAD_ID, + title: "Search fixture", + agent_name: "researcher", + }, + ], + }); +}); + +test("keyboard commands and Escape work without a search provider", async ({ + page, +}) => { + await page.route("**/api/search", (route) => + route.fulfill({ + json: { + items: [], + cursor: null, + indexing: false, + messages_available: false, + }, + }), + ); + await page.goto("/workspace/chats/new"); + await expect( + page.getByRole("textbox", { name: "How can I assist you today?" }), + ).toBeVisible({ timeout: 15_000 }); + await expect( + page.getByRole("button", { name: /^Search/ }).first(), + ).toBeVisible(); + await page.keyboard.press("ControlOrMeta+k"); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + const input = dialog.getByPlaceholder("Search titles and messages…"); + await expect(input).toBeFocused(); + await input.fill(">settings"); + await expect(dialog.getByRole("option", { name: /Settings/ })).toBeVisible(); + await input.press("Enter"); + await expect(dialog).toBeHidden(); + await expect(page.getByRole("dialog")).toBeVisible(); + await page.keyboard.press("Escape"); + await expect(page.getByRole("dialog")).toBeHidden(); +}); + +test("filters, stale requests, safe snippets and custom agent links", async ({ + page, +}) => { + const requests: { + query: string; + types: string[]; + archived: boolean | null; + }[] = []; + await page.route("**/api/search", async (route) => { + const body = route.request().postDataJSON() as (typeof requests)[number]; + requests.push(body); + if (body.query === "old") + await new Promise((resolve) => setTimeout(resolve, 700)); + await route.fulfill({ + json: { + items: body.query + ? [ + { + type: "message", + thread_id: MOCK_THREAD_ID, + title: `${body.query} result`, + agent_name: "researcher", + archived: true, + updated_at: "2026-10-25T08:00:00Z", + project_id: null, + message_id: "original/id", + seq: 3, + role: "assistant", + snippet: [ + { text: ``, matched: true }, + ], + }, + ] + : [], + cursor: null, + indexing: false, + messages_available: true, + }, + }); + }); + await page.goto("/workspace/chats/new"); + await expect( + page.getByRole("textbox", { name: "How can I assist you today?" }), + ).toBeVisible({ timeout: 15_000 }); + await page + .getByRole("button", { name: /^Search/ }) + .first() + .click(); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + const input = dialog.getByPlaceholder("Search titles and messages…"); + await input.fill("old"); + await expect.poll(() => requests.some((r) => r.query === "old")).toBeTruthy(); + await input.fill("latest"); + await expect(dialog.getByText("latest result")).toBeVisible(); + await expect(dialog.locator("mark")).toHaveText(""); + await dialog.getByRole("button", { name: "Messages", exact: true }).click(); + await dialog.getByRole("checkbox", { name: "Include archived" }).uncheck(); + await expect + .poll(() => requests.at(-1)) + .toMatchObject({ query: "latest", types: ["message"], archived: false }); + await expect(dialog.getByText("old result")).toBeHidden(); + await dialog.getByRole("option").filter({ hasText: "latest result" }).click(); + await expect(page).toHaveURL( + new RegExp( + `/workspace/agents/researcher/chats/${MOCK_THREAD_ID}\\?message=original%2Fid&message_seq=3`, + ), + ); +}); + +test("empty and error states retry without losing the query", async ({ + page, +}) => { + let fail = true; + await page.route("**/api/search", (route) => + fail + ? route.fulfill({ status: 503, json: {} }) + : route.fulfill({ + json: { + items: [], + cursor: null, + indexing: false, + messages_available: true, + }, + }), + ); + await page.goto("/workspace/chats/new"); + await expect( + page.getByRole("textbox", { name: "How can I assist you today?" }), + ).toBeVisible({ timeout: 15_000 }); + await page + .getByRole("button", { name: /^Search/ }) + .first() + .click(); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + const input = dialog.getByPlaceholder("Search titles and messages…"); + await input.fill("missing report"); + await expect( + dialog.getByText("Search could not finish. Try again."), + ).toBeVisible(); + fail = false; + await dialog.getByRole("button", { name: "Try again" }).click(); + await expect(dialog.getByText(/No matches/)).toBeVisible(); + await expect(input).toHaveValue("missing report"); +}); + +test("Chinese composition does not dispatch partial searches or select a command", async ({ + page, +}) => { + const queries: string[] = []; + await page.route("**/api/search", (route) => { + queries.push((route.request().postDataJSON() as { query: string }).query); + return route.fulfill({ + json: { + items: [], + cursor: null, + indexing: false, + messages_available: true, + }, + }); + }); + await page.goto("/workspace/chats/new"); + await expect( + page.getByRole("textbox", { name: "How can I assist you today?" }), + ).toBeVisible({ timeout: 15_000 }); + await page + .getByRole("button", { name: /^Search/ }) + .first() + .click(); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + const input = dialog.getByPlaceholder("Search titles and messages…"); + await input.dispatchEvent("compositionstart"); + await input.fill("baogao"); + await input.dispatchEvent("keydown", { + key: "Enter", + keyCode: 229, + isComposing: true, + }); + await expect(dialog).toBeVisible(); + await page.waitForTimeout(350); + expect(queries).not.toContain("baogao"); + await input.fill("报告"); + await input.dispatchEvent("compositionend"); + await expect.poll(() => queries).toContain("报告"); +}); + +test("search fits a narrow viewport", async ({ page }) => { + await page.setViewportSize({ width: 390, height: 844 }); + await page.route("**/api/search", (route) => + route.fulfill({ + json: { + items: [], + cursor: null, + indexing: false, + messages_available: true, + }, + }), + ); + await page.goto("/workspace/chats/new"); + await expect( + page.getByRole("textbox", { name: "How can I assist you today?" }), + ).toBeVisible({ timeout: 15_000 }); + await page.getByRole("button", { name: "Toggle Sidebar" }).first().click(); + await page + .getByRole("button", { name: /^Search/ }) + .first() + .click(); + const dialog = page.getByRole("dialog", { name: "Search conversations" }); + await expect(dialog).toBeVisible(); + const bounds = await dialog.boundingBox(); + expect(bounds!.x).toBeGreaterThanOrEqual(0); + expect(bounds!.x + bounds!.width).toBeLessThanOrEqual(390); + await expect( + dialog.getByRole("checkbox", { name: "Include archived" }), + ).toBeInViewport(); + await page.keyboard.press("Escape"); + await expect(dialog).toBeHidden(); +}); diff --git a/frontend/tests/unit/core/search/hooks.dom.test.tsx b/frontend/tests/unit/core/search/hooks.dom.test.tsx new file mode 100644 index 0000000000..d404bbe658 --- /dev/null +++ b/frontend/tests/unit/core/search/hooks.dom.test.tsx @@ -0,0 +1,58 @@ +import { afterEach, expect, rs, test } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { cleanup, renderHook, waitFor } from "@testing-library/react"; +import type { ReactNode } from "react"; + +import { useWorkspaceSearch } from "@/core/search/hooks"; + +let viewer = { id: "alice", permissions: ["threads:read"] }; +rs.mock("@/core/auth/AuthProvider", () => ({ + useAuth: () => ({ user: viewer }), +})); +const search = rs.fn(); +rs.mock("@/core/search/api", () => ({ + searchWorkspace: (...args: unknown[]) => search(...args), +})); +rs.mock("@/core/static-mode", () => ({ isStaticWebsiteOnly: () => false })); +afterEach(() => { + cleanup(); + rs.clearAllMocks(); + viewer = { id: "alice", permissions: ["threads:read"] }; +}); + +test("debounce retains results only within the same owner, permission and filter scope", async () => { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + search.mockResolvedValueOnce({ + items: [{ type: "thread", thread_id: "old" }], + cursor: null, + }); + search.mockImplementation(() => new Promise(() => undefined)); + const { result, rerender } = renderHook( + ({ query, archived }) => + useWorkspaceSearch( + query, + { type: "thread", includeArchived: archived }, + true, + ), + { wrapper, initialProps: { query: "old", archived: false } }, + ); + await waitFor(() => expect(result.current.items).toHaveLength(1)); + rerender({ query: "new", archived: false }); + expect(result.current.waiting).toBe(true); + expect(result.current.items).toHaveLength(1); + rerender({ query: "new", archived: true }); + expect(result.current.items).toHaveLength(0); + viewer = { id: "bob", permissions: ["threads:read"] }; + rerender({ query: "old", archived: false }); + expect(result.current.items).toHaveLength(0); + viewer = { id: "alice", permissions: [] }; + rerender({ query: "old", archived: false }); + expect(result.current.supported).toBe(false); + expect(result.current.items).toHaveLength(0); + client.clear(); +});