Merge branch 'ji'

2026-06-11 19:59:58 +08:00
commit 82600a3404
@@ -250,7 +250,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
 | **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
 | **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
-| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` → `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "已改用规则抽取". The "保存到知识库" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/file` - **import an uploaded file** (multipart: `file` + optional `title`/`tags`/`status`/`model_name`) into wiki note(s): the file is converted to Markdown (md/txt read directly, pdf/word/ppt/excel via `convert_file_to_markdown`), distilled by the same wiki-capture LLM pipeline as thread capture (`source_type="file"`), and falls back to a verbatim **draft** note when the model returns nothing usable (`llm_fallback:true`); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector→keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[…]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). Backed by the `deerflow.knowledge` module (see below). |
+| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` → `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "已改用规则抽取". The "保存到知识库" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/file` - **import an uploaded file** (multipart: `file` + optional `title`/`tags`/`status`/`model_name`) into wiki note(s): the file is converted to Markdown (md/txt read directly, pdf/word/ppt/excel via `convert_file_to_markdown`), distilled by the same wiki-capture LLM pipeline as thread capture (`source_type="file"`), and falls back to a verbatim **draft** note when the model returns nothing usable (`llm_fallback:true`); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector→keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities, graph-blacklist filtered); `GET/POST /graph/blacklist` + `DELETE /graph/blacklist/{id}` - graph node blacklist (labels = entity names / note titles, case-insensitive match; blacklisted nodes and their edges are dropped from `/graph` and the wiki-export artifacts; table `knowledge_graph_blacklist`, auto-created); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[…]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). Backed by the `deerflow.knowledge` module (see below). |

 Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.

@@ -864,9 +864,12 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
                tool_groups=next_tool_groups,
                common_questions=next_common_questions,
            )
-            # Keep the listing cache in lock-step with config.yaml on every write.
-            await store.replace_skills(agent_id, next_skills)
-            await store.set_extras(agent_id, tool_groups=next_tool_groups, model=next_model)
+            # Keep the listing cache in lock-step with config.yaml, but only
+            # when the cached fields were actually touched — a partial update
+            # (e.g. 常用问题 only) skips the DELETE/INSERT churn.
+            if any(key in fields_set for key in ("skills", "model", "tool_groups")):
+                await store.replace_skills(agent_id, next_skills)
+                await store.set_extras(agent_id, tool_groups=next_tool_groups, model=next_model)
        if request.soul is not None:
            _write_agent_soul(agent_id, request.soul)

@@ -571,6 +571,37 @@ async def get_graph(request: Request) -> dict[str, Any]:
    return await service.get_graph()


+class GraphBlacklistAddRequest(BaseModel):
+    labels: list[str] = Field(default_factory=list)
+
+
+@router.get("/graph/blacklist")
+@require_auth
+async def list_graph_blacklist(request: Request) -> dict[str, Any]:
+    """Return blacklisted graph node labels (hidden from the graph)."""
+    service = get_knowledge_service(request)
+    return {"items": await service.list_graph_blacklist()}
+
+
+@router.post("/graph/blacklist")
+@require_auth
+async def add_graph_blacklist(body: GraphBlacklistAddRequest, request: Request) -> dict[str, Any]:
+    """Blacklist node labels (entity names / note titles); they disappear from the graph."""
+    service = get_knowledge_service(request)
+    added = await service.add_graph_blacklist(body.labels, created_by=get_effective_user_id())
+    return {"added": added, "items": await service.list_graph_blacklist()}
+
+
+@router.delete("/graph/blacklist/{entry_id}")
+@require_auth
+async def remove_graph_blacklist(entry_id: str, request: Request) -> dict[str, Any]:
+    """Remove a blacklist entry so the node shows up in the graph again."""
+    service = get_knowledge_service(request)
+    if not await service.remove_graph_blacklist(entry_id):
+        raise HTTPException(status_code=404, detail="Blacklist entry not found")
+    return {"ok": True}
+
+
@router.post("/export/graph")
@require_auth
 async def export_graph(request: Request) -> dict[str, Any]:
@@ -173,6 +173,7 @@ class ThreadSearchRequest(BaseModel):
    limit: int = Field(default=100, ge=1, le=1000, description="Maximum results")
    offset: int = Field(default=0, ge=0, description="Pagination offset")
    status: str | None = Field(default=None, description="Filter by thread status")
+    query: str | None = Field(default=None, description="Case-insensitive substring match on the thread title")


 class ThreadStateResponse(BaseModel):
@@ -660,11 +661,19 @@ async def search_threads(body: ThreadSearchRequest, request: Request) -> list[Th
    from app.gateway.deps import get_thread_store

    repo = get_thread_store(request)
+    # System threads (scheduler runs, roundtable workers, …) carry
+    # ``metadata.system=true`` and are machine-created — hide them from the
+    # default conversation list so they don't crowd out real chats. A caller
+    # that explicitly filters on ``thread_type``/``system`` (e.g. the studio
+    # notebook list) opts out of the exclusion.
+    explicit_system_filter = bool(body.metadata) and ("thread_type" in body.metadata or "system" in body.metadata)
    rows = await repo.search(
        metadata=body.metadata or None,
        status=body.status,
        limit=body.limit,
        offset=body.offset,
+        exclude_system=not explicit_system_filter,
+        query=body.query,
    )
    return [
        ThreadResponse(
@@ -25,10 +25,22 @@ def _vault_key(vault_path: str | None) -> str | None:
    return key


+def normalize_blacklist(labels: list[str] | set[str] | None) -> set[str]:
+    """Normalize blacklist labels for matching (trimmed, lower-cased, non-empty)."""
+    out: set[str] = set()
+    for raw in labels or []:
+        label = (raw or "").strip().lower()
+        if label:
+            out.add(label)
+    return out
+
+
 def build_graph(
    notes: list[dict[str, Any]],
    entities: list[dict[str, Any]] | None = None,
    relations: list[dict[str, Any]] | None = None,
+    *,
+    blacklist: set[str] | None = None,
 ) -> dict[str, Any]:
    """Return ``{nodes, edges}`` from notes (+ optional entities/relations).

@@ -36,7 +48,20 @@ def build_graph(
    when ``entities``/``relations`` are supplied (phase 4), entity nodes and
    note→entity / entity→entity edges are added on top of the wikilink and
    shared-tag edges.
+
+    ``blacklist`` is a set of *normalized* labels (see :func:`normalize_blacklist`);
+    notes whose title and entities whose name match are dropped, together with
+    every edge touching them (edges only join surviving nodes).
    """
+    blocked = blacklist or set()
+
+    def _is_blocked(label: str | None) -> bool:
+        return bool(blocked) and (label or "").strip().lower() in blocked
+
+    # Drop blacklisted notes up front so the wikilink / shared-tag edge loops
+    # below never reference them either.
+    notes = [n for n in notes if not _is_blocked(n.get("title"))]
+
    nodes = []
    by_key: dict[str, str] = {}
    note_ids: set[str] = set()
@@ -59,6 +84,8 @@ def build_graph(
    for ent in entities or []:
        if ent["id"] not in mentioned:
            continue
+        if _is_blocked(ent.get("name")):
+            continue
        nodes.append({"id": ent["id"], "type": "entity", "label": ent.get("name") or ent["id"], "category": ent.get("entity_type")})
        entity_ids.add(ent["id"])

@@ -156,11 +183,12 @@ def export_all(
    relations: list[dict[str, Any]] | None = None,
    write_html: bool = True,
    write_json: bool = True,
+    blacklist: set[str] | None = None,
 ) -> dict[str, Any]:
    """Generate graph artifacts under ``<vault>/wiki-export/`` and return stats."""
    out_dir = Path(vault_root).resolve() / "wiki-export"
    out_dir.mkdir(parents=True, exist_ok=True)
-    graph = build_graph(notes, entities, relations)
+    graph = build_graph(notes, entities, relations, blacklist=blacklist)
    written: list[str] = []
    if write_json:
        (out_dir / "graph.json").write_text(json.dumps(graph, ensure_ascii=False, indent=2), encoding="utf-8")
@@ -126,6 +126,21 @@ class KnowledgeEntityRow(Base):
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))


+class KnowledgeGraphBlacklistRow(Base):
+    """Graph blacklist: node labels (entity names / note titles) hidden from the graph.
+
+    Matching is case-insensitive on the trimmed label. Global/shared like the
+    rest of the knowledge base; ``created_by`` is audit-only.
+    """
+
+    __tablename__ = "knowledge_graph_blacklist"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    label: Mapped[str] = mapped_column(String(512), unique=True, index=True)
+    created_by: Mapped[str | None] = mapped_column(String(128))
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
+
+
 class KnowledgeRelationRow(Base):
    """Relation between notes/entities (phase 4, mirrors ``knowledge_relations``)."""

@@ -150,6 +150,7 @@ class ObsidianWikiAdapter:
        entities: list[dict[str, Any]] | None = None,
        relations: list[dict[str, Any]] | None = None,
        updated: str,
+        blacklist: set[str] | None = None,
    ) -> dict[str, Any]:
        self.ensure_ready()
        result = exporter.export_all(
@@ -159,6 +160,7 @@ class ObsidianWikiAdapter:
            relations=relations,
            write_html=self.export_html,
            write_json=self.export_json,
+            blacklist=blacklist,
        )
        self._append_log("EXPORT", "graph", f"nodes={result['stats']['nodes']} edges={result['stats']['edges']}", updated)
        return result
@@ -18,6 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from deerflow.knowledge.models import (
    KnowledgeEmbeddingRow,
    KnowledgeEntityRow,
+    KnowledgeGraphBlacklistRow,
    KnowledgeNoteRow,
    KnowledgeNoteTagRow,
    KnowledgeNoteVersionRow,
@@ -485,6 +486,49 @@ class KnowledgeRepository:
            rows = (await session.execute(select(KnowledgeEntityRow))).scalars()
            return [{"id": r.id, "name": r.name, "entity_type": r.entity_type} for r in rows]

+    # -- graph blacklist ---------------------------------------------------
+    async def list_graph_blacklist(self) -> list[dict[str, Any]]:
+        """Return every blacklisted graph node label, newest first."""
+        async with self._sf() as session:
+            stmt = select(KnowledgeGraphBlacklistRow).order_by(KnowledgeGraphBlacklistRow.created_at.desc())
+            rows = (await session.execute(stmt)).scalars()
+            return [{"id": r.id, "label": r.label, "created_by": r.created_by, "created_at": _iso(r.created_at)} for r in rows]
+
+    async def add_graph_blacklist(self, labels: list[str], *, created_by: str | None = None) -> int:
+        """Add labels to the graph blacklist (trimmed, case-insensitive dedup). Returns the count actually added."""
+        cleaned: list[str] = []
+        seen: set[str] = set()
+        for raw in labels or []:
+            label = (raw or "").strip()
+            if not label or label.lower() in seen:
+                continue
+            seen.add(label.lower())
+            cleaned.append(label)
+        if not cleaned:
+            return 0
+        async with self._sf() as session:
+            existing = {
+                value
+                for (value,) in (
+                    await session.execute(select(func.lower(KnowledgeGraphBlacklistRow.label)).where(func.lower(KnowledgeGraphBlacklistRow.label).in_([label.lower() for label in cleaned])))
+                ).all()
+            }
+            added = 0
+            for label in cleaned:
+                if label.lower() in existing:
+                    continue
+                session.add(KnowledgeGraphBlacklistRow(id=_new_id("kbblk"), label=label, created_by=created_by, created_at=datetime.now(UTC)))
+                added += 1
+            if added:
+                await session.commit()
+            return added
+
+    async def remove_graph_blacklist(self, entry_id: str) -> bool:
+        async with self._sf() as session:
+            result = await session.execute(delete(KnowledgeGraphBlacklistRow).where(KnowledgeGraphBlacklistRow.id == entry_id))
+            await session.commit()
+            return bool(result.rowcount)
+
    async def list_relations(self) -> list[dict[str, Any]]:
        async with self._sf() as session:
            rows = (await session.execute(select(KnowledgeRelationRow))).scalars()
@@ -796,7 +796,8 @@ class KnowledgeService:
        notes = await self.repo.list_all()
        entities = await self.repo.list_entities() if self.config.entity_extraction_enabled else []
        relations = await self.repo.list_relations() if self.config.entity_extraction_enabled else []
-        result = self.adapter.export_graph(notes, entities=entities, relations=relations, updated=now_iso())
+        blacklist = await self._graph_blacklist_set()
+        result = self.adapter.export_graph(notes, entities=entities, relations=relations, updated=now_iso(), blacklist=blacklist)
        return {"files": result["files"], "stats": result["stats"]}

    async def get_graph(self) -> dict[str, Any]:
@@ -806,7 +807,26 @@ class KnowledgeService:
        notes = await self.repo.list_all()
        entities = await self.repo.list_entities() if self.config.entity_extraction_enabled else []
        relations = await self.repo.list_relations() if self.config.entity_extraction_enabled else []
-        return exporter.build_graph(notes, entities, relations)
+        blacklist = await self._graph_blacklist_set()
+        return exporter.build_graph(notes, entities, relations, blacklist=blacklist)
+
+    # -- graph blacklist ---------------------------------------------------
+    async def _graph_blacklist_set(self) -> set[str]:
+        from deerflow.knowledge import exporter
+
+        items = await self.repo.list_graph_blacklist()
+        return exporter.normalize_blacklist([item["label"] for item in items])
+
+    async def list_graph_blacklist(self) -> list[dict[str, Any]]:
+        """Return blacklist entries (node labels hidden from the graph)."""
+        return await self.repo.list_graph_blacklist()
+
+    async def add_graph_blacklist(self, labels: list[str], *, created_by: str | None = None) -> int:
+        """Blacklist node labels (entity names / note titles). Returns the count added."""
+        return await self.repo.add_graph_blacklist(labels, created_by=created_by)
+
+    async def remove_graph_blacklist(self, entry_id: str) -> bool:
+        return await self.repo.remove_graph_blacklist(entry_id)

    def export_file_path(self, filename: str) -> Path | None:
        return self.adapter.export_file_path(filename)
@@ -28,6 +28,7 @@ from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.knowledge.models import (
    KnowledgeEmbeddingRow,
    KnowledgeEntityRow,
+    KnowledgeGraphBlacklistRow,
    KnowledgeNoteRow,
    KnowledgeNoteTagRow,
    KnowledgeNoteVersionRow,
@@ -75,6 +76,7 @@ __all__ = [
    "BusinessMappingRow",
    "KnowledgeEmbeddingRow",
    "KnowledgeEntityRow",
+    "KnowledgeGraphBlacklistRow",
    "KnowledgeNoteRow",
    "KnowledgeNoteTagRow",
    "KnowledgeNoteVersionRow",
@@ -45,7 +45,18 @@ class ThreadMetaStore(abc.ABC):
        limit: int = 100,
        offset: int = 0,
        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
    ) -> list[dict]:
+        """Search threads ordered by ``updated_at`` desc.
+
+        ``exclude_system=True`` drops threads whose ``metadata.system`` is
+        truthy (scheduler runs, roundtable workers, …) so user-facing
+        conversation lists are not crowded out by machine-created threads.
+        ``query`` is a case-insensitive substring match on the thread title
+        (``display_name``). ``limit``/``offset`` apply to the *filtered*
+        result set.
+        """
        pass

    @abc.abstractmethod
@@ -74,6 +74,8 @@ class MemoryThreadMetaStore(ThreadMetaStore):
        limit: int = 100,
        offset: int = 0,
        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
    ) -> list[dict]:
        resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.search")
        filter_dict: dict[str, Any] = {}
@@ -84,13 +86,45 @@ class MemoryThreadMetaStore(ThreadMetaStore):
        if resolved_user_id is not None:
            filter_dict["user_id"] = resolved_user_id

-        items = await self._store.asearch(
-            THREADS_NS,
-            filter=filter_dict or None,
-            limit=limit,
-            offset=offset,
-        )
-        return [self._item_to_dict(item) for item in items]
+        normalized_query = (query or "").strip().lower()
+        if not exclude_system and not normalized_query:
+            items = await self._store.asearch(
+                THREADS_NS,
+                filter=filter_dict or None,
+                limit=limit,
+                offset=offset,
+            )
+            return [self._item_to_dict(item) for item in items]
+
+        # The Store filter only supports equality, so the system-thread
+        # exclusion and title search are applied in Python: scan in batches
+        # until the requested window is full so ``limit``/``offset`` refer to
+        # filtered rows.
+        def _matches(record: dict) -> bool:
+            if exclude_system and (record.get("metadata") or {}).get("system"):
+                return False
+            if normalized_query and normalized_query not in (record.get("display_name") or "").lower():
+                return False
+            return True
+
+        needed = offset + limit
+        batch_size = max(500, needed)
+        matched: list[dict] = []
+        scan_offset = 0
+        while len(matched) < needed:
+            items = await self._store.asearch(
+                THREADS_NS,
+                filter=filter_dict or None,
+                limit=batch_size,
+                offset=scan_offset,
+            )
+            if not items:
+                break
+            matched.extend(record for record in (self._item_to_dict(item) for item in items) if _matches(record))
+            if len(items) < batch_size:
+                break
+            scan_offset += len(items)
+        return matched[offset:needed]

    async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
        item = await self._store.aget(THREADS_NS, thread_id)
@@ -5,7 +5,7 @@ from __future__ import annotations
 from datetime import UTC, datetime
 from typing import Any

-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

 from deerflow.persistence.thread_meta.base import ThreadMetaStore
@@ -109,6 +109,8 @@ class ThreadMetaRepository(ThreadMetaStore):
        limit: int = 100,
        offset: int = 0,
        user_id: str | None | _AutoSentinel = AUTO,
+        exclude_system: bool = False,
+        query: str | None = None,
    ) -> list[dict]:
        """Search threads with optional metadata and status filters.

@@ -116,28 +118,55 @@ class ThreadMetaRepository(ThreadMetaStore):
        context. Pass ``user_id=None`` to bypass (migration/CLI).
        """
        resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.search")
-        stmt = select(ThreadMetaRow).order_by(ThreadMetaRow.updated_at.desc())
+        # thread_id tie-breaker keeps OFFSET pagination stable when many rows
+        # share the same updated_at (bulk imports, scheduler bursts).
+        stmt = select(ThreadMetaRow).order_by(ThreadMetaRow.updated_at.desc(), ThreadMetaRow.thread_id.desc())
        if resolved_user_id is not None:
            stmt = stmt.where(ThreadMetaRow.user_id == resolved_user_id)
        if status:
            stmt = stmt.where(ThreadMetaRow.status == status)
+        if query and query.strip():
+            # Case-insensitive title search. display_name is a real column, so
+            # this filter runs in SQL even when the JSON filters below fall
+            # back to Python-side scanning.
+            stmt = stmt.where(func.lower(ThreadMetaRow.display_name).contains(query.strip().lower()))

-        if metadata:
-            # When metadata filter is active, fetch a larger window and filter
-            # in Python. TODO(Phase 2): use JSON DB operators (Postgres @>,
-            # SQLite json_extract) for server-side filtering.
-            stmt = stmt.limit(limit * 5 + offset)
-            async with self._sf() as session:
-                result = await session.execute(stmt)
-                rows = [self._row_to_dict(r) for r in result.scalars()]
-            rows = [r for r in rows if all(r.get("metadata", {}).get(k) == v for k, v in metadata.items())]
-            return rows[offset : offset + limit]
-        else:
+        if not metadata and not exclude_system:
            stmt = stmt.limit(limit).offset(offset)
            async with self._sf() as session:
                result = await session.execute(stmt)
                return [self._row_to_dict(r) for r in result.scalars()]

+        # Metadata filters live inside the JSON blob, so rows are scanned in
+        # batches and filtered in Python until the requested window is full.
+        # A fixed over-fetch window would under-fill pages when most rows are
+        # filtered out (e.g. heavy scheduled-task usage), making callers that
+        # paginate believe the list is exhausted. TODO(Phase 2): use JSON DB
+        # operators (Postgres @>, SQLite json_extract) for server-side filtering.
+        def _matches(row: dict) -> bool:
+            md = row.get("metadata") or {}
+            if metadata and not all(md.get(k) == v for k, v in metadata.items()):
+                return False
+            if exclude_system and md.get("system"):
+                return False
+            return True
+
+        needed = offset + limit
+        batch_size = max(500, needed)
+        matched: list[dict] = []
+        scan_offset = 0
+        async with self._sf() as session:
+            while len(matched) < needed:
+                result = await session.execute(stmt.limit(batch_size).offset(scan_offset))
+                rows = result.scalars().all()
+                if not rows:
+                    break
+                matched.extend(d for d in (self._row_to_dict(r) for r in rows) if _matches(d))
+                if len(rows) < batch_size:
+                    break
+                scan_offset += len(rows)
+        return matched[offset:needed]
+
    async def _check_ownership(self, session: AsyncSession, thread_id: str, resolved_user_id: str | None) -> bool:
        """Return True if the row exists and is owned (or filter bypassed)."""
        if resolved_user_id is None: