fix: restore embedded wiki images and optimize vector sync

offline-backend-20260512/backend/packages/harness/deerflow/config/llmwiki_config.py

@@ -59,7 +59,7 @@ class LocalWikiEmbeddingConfig(BaseModel):
     base_url: str = ""
     api_key: str = Field(default="", repr=False)
     dimensions: int | None = Field(
-        default=1024,
+        default=None,
         ge=1,
         le=65536,
         description="Expected response vector dimensions; not sent to the embedding endpoint.",
@@ -108,9 +108,7 @@ class LocalWikiIndexConfig(BaseModel):
     """DeerFlow-owned mirror and exact-vector index for WeKnora Wiki pages."""
 
     enabled: bool = False
-    sync_enabled: bool = False
-    internal_search_enabled: bool = False
-    reference_ui_enabled: bool = False
+    auto_sync: bool = False
     sync_interval_seconds: int = Field(default=120, ge=10, le=86400)
     sync_page_size: int = Field(default=100, ge=1, le=500)
     sync_concurrency: int = Field(default=3, ge=1, le=32)
@@ -133,9 +131,31 @@ class LocalWikiIndexConfig(BaseModel):
     keyword_boost_weight: float = Field(default=0.15, ge=0, le=1)
     cache_max_knowledge_bases: int = Field(default=20, ge=1, le=1000)
     cache_revision_check_seconds: float = Field(default=5, ge=0, le=3600)
+    database_batch_size: int = Field(default=50, ge=1, le=500)
     external_api: LocalWikiExternalApiConfig = Field(default_factory=LocalWikiExternalApiConfig)
     model_config = ConfigDict(extra="ignore")
 
+    @model_validator(mode="before")
+    @classmethod
+    def _migrate_legacy_flags(cls, value):
+        if isinstance(value, dict) and "auto_sync" not in value and "sync_enabled" in value:
+            value = {**value, "auto_sync": value.get("sync_enabled")}
+        return value
+
+    @property
+    def sync_enabled(self) -> bool:
+        """Backward-compatible status name; automatic work is one explicit switch."""
+
+        return self.auto_sync
+
+    @property
+    def internal_search_enabled(self) -> bool:
+        return self.enabled
+
+    @property
+    def reference_ui_enabled(self) -> bool:
+        return self.enabled
+
     @model_validator(mode="after")
     def _validate_index(self):
         if self.chunk_overlap_chars >= self.chunk_max_chars:

offline-backend-20260512/backend/packages/harness/deerflow/integrations/weknora/client.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
 import time
+from collections.abc import Mapping, Sequence
 from typing import Any
 from urllib.parse import quote
 
@@ -430,6 +430,51 @@ class WeKnoraClient:
         media_type = response.headers.get("content-type", "application/octet-stream").split(";", 1)[0].strip()
         return response.content, media_type or "application/octet-stream"
 
+    async def get_knowledge_base_file(
+        self,
+        knowledge_base_id: str,
+        file_path: str,
+    ) -> tuple[bytes, str]:
+        """Return one protected Wiki attachment using DeerFlow's WeKnora identity."""
+
+        owns_client = self._http_client is None
+        client = self._http_client or httpx.AsyncClient(
+            timeout=httpx.Timeout(self.timeout_seconds),
+            follow_redirects=True,
+            trust_env=False,
+        )
+        try:
+            for attempt in range(2):
+                response = await client.request(
+                    "GET",
+                    f"{self.base_url}/api/v1/knowledge-bases/{quote(knowledge_base_id, safe='')}/files",
+                    params={"file_path": file_path},
+                    headers=await self._headers(force_refresh=attempt > 0, accept="*/*"),
+                    follow_redirects=True,
+                )
+                if response.status_code not in {401, 403} or not self._has_admin_credentials or attempt > 0:
+                    break
+        except httpx.TimeoutException as exc:
+            raise WeKnoraError("WeKnora Wiki image request timed out", status_code=504) from exc
+        except httpx.HTTPError as exc:
+            raise WeKnoraError("WeKnora Wiki image is unavailable", status_code=502) from exc
+        finally:
+            if owns_client:
+                await client.aclose()
+        if response.status_code == 404:
+            raise WeKnoraError("WeKnora Wiki image was not found", status_code=404)
+        if response.status_code in {401, 403}:
+            if self._has_admin_credentials:
+                raise WeKnoraError("WeKnora rejected DeerFlow's authenticated Wiki image request", status_code=502)
+            raise WeKnoraError(
+                "WeKnora rejected DeerFlow's anonymous internal request; allow DeerFlow in WeKnora or its reverse proxy",
+                status_code=502,
+            )
+        if response.status_code >= 400:
+            raise WeKnoraError(f"WeKnora returned HTTP {response.status_code}", status_code=502)
+        media_type = response.headers.get("content-type", "application/octet-stream").split(";", 1)[0].strip()
+        return response.content, media_type or "application/octet-stream"
+
     async def reprocess_document(self, knowledge_id: str) -> dict[str, Any]:
         value = _data(await self._request("POST", f"/api/v1/knowledge/{quote(knowledge_id, safe='')}/reparse"))
         return dict(value) if isinstance(value, dict) else {}

offline-backend-20260512/backend/packages/harness/deerflow/integrations/weknora/local_index/embedding.py

@@ -15,6 +15,21 @@ from deerflow.integrations.weknora.local_index.vector_codec import normalize_vec
 logger = logging.getLogger(__name__)
 
 
+def _upstream_error_detail(response: httpx.Response) -> str:
+    """Return a bounded upstream reason without hiding the useful response body."""
+
+    try:
+        payload = response.json()
+        if isinstance(payload, dict):
+            detail = payload.get("detail") or payload.get("message") or payload.get("error") or payload
+        else:
+            detail = payload
+        value = str(detail)
+    except (ValueError, TypeError):
+        value = response.text
+    return " ".join(value.split())[:2000] or "no response body"
+
+
 class WikiEmbeddingError(RuntimeError):
     def __init__(self, message: str, *, category: str = "request", retryable: bool = True) -> None:
         super().__init__(message)
@@ -76,11 +91,15 @@ class StrictWikiEmbeddingClient:
             started = monotonic()
             try:
                 response = await (await self._get_client()).post(url, headers=headers, json=payload)
-                if response.status_code in {401, 403}:
-                    raise WikiEmbeddingError("Embedding authentication failed", category="authentication", retryable=False)
-                if response.status_code == 429 or response.status_code >= 500:
-                    raise WikiEmbeddingError(f"Embedding service returned HTTP {response.status_code}", category="upstream", retryable=True)
-                response.raise_for_status()
+                if response.status_code >= 400:
+                    detail = _upstream_error_detail(response)
+                    category = "authentication" if response.status_code in {401, 403} else "upstream"
+                    retryable = response.status_code == 429 or response.status_code >= 500
+                    raise WikiEmbeddingError(
+                        f"Embedding service returned HTTP {response.status_code}: {detail}",
+                        category=category,
+                        retryable=retryable,
+                    )
                 data = response.json()
                 items = sorted(data.get("data", []), key=lambda item: int(item.get("index", 0))) if isinstance(data, dict) else []
                 if len(items) != len(texts):
@@ -98,8 +117,20 @@ class StrictWikiEmbeddingClient:
             except httpx.TimeoutException:
                 last_error = WikiEmbeddingError("Embedding request timed out", category="timeout", retryable=True)
             except (httpx.HTTPError, ValueError, TypeError) as exc:
-                last_error = WikiEmbeddingError(f"Embedding request failed ({type(exc).__name__})", category="request", retryable=isinstance(exc, httpx.HTTPError))
+                reason = " ".join(str(exc).split())[:2000] or type(exc).__name__
+                last_error = WikiEmbeddingError(
+                    f"Embedding request failed ({type(exc).__name__}): {reason}",
+                    category="request",
+                    retryable=isinstance(exc, httpx.HTTPError),
+                )
             if last_error is not None and (not last_error.retryable or attempt >= self.config.max_retries):
+                logger.error(
+                    "Wiki embedding batch failed model=%s batch=%d category=%s reason=%s",
+                    self.config.model,
+                    len(texts),
+                    last_error.category,
+                    last_error,
+                )
                 raise last_error
             await asyncio.sleep(min(30.0, 0.5 * (2**attempt)))
         raise last_error or WikiEmbeddingError("Embedding request failed")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/llmwiki_index/__init__.py

@@ -6,8 +6,12 @@ from deerflow.persistence.llmwiki_index.model import LlmWikiWikiPageRow, LlmWiki
 from deerflow.persistence.llmwiki_index.sql import SqlLlmWikiIndexStore
 
 
-def make_llmwiki_index_store(session_factory):
-    return MemoryLlmWikiIndexStore() if session_factory is None else SqlLlmWikiIndexStore(session_factory)
+def make_llmwiki_index_store(session_factory, *, write_batch_size: int = 50):
+    return (
+        MemoryLlmWikiIndexStore()
+        if session_factory is None
+        else SqlLlmWikiIndexStore(session_factory, write_batch_size=write_batch_size)
+    )
 
 
 __all__ = [

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

@@ -6,7 +6,7 @@ from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import uuid4
 
-from sqlalchemy import case, delete, func, or_, select, update
+from sqlalchemy import case, delete, func, insert, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -27,8 +27,9 @@ def _dict(row) -> dict[str, Any]:
 
 
 class SqlLlmWikiIndexStore(LlmWikiIndexStore):
-    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession], *, write_batch_size: int = 50) -> None:
         self._sf = session_factory
+        self.write_batch_size = max(1, min(int(write_batch_size), 500))
 
     async def _ensure_state(self, session: AsyncSession, mapping_id: str) -> LlmWikiWikiSyncStateRow:
         row = await session.get(LlmWikiWikiSyncStateRow, mapping_id)
@@ -82,24 +83,29 @@ class SqlLlmWikiIndexStore(LlmWikiIndexStore):
             row.updated_at = now
             await session.flush()
             await session.execute(delete(LlmWikiWikiVectorRow).where(LlmWikiWikiVectorRow.page_id == row.id))
-            session.add_all(
-                [
-                    LlmWikiWikiVectorRow(
-                        id=str(uuid4()),
-                        page_id=row.id,
-                        knowledge_base_mapping_id=mapping_id,
-                        section_index=int(item["section_index"]),
-                        heading=item.get("heading"),
-                        section_content=str(item["section_content"]),
-                        content_hash=str(item["content_hash"]),
-                        embedding_fingerprint=fingerprint,
-                        embedding_dimensions=int(item["embedding_dimensions"]),
-                        vector_blob=bytes(item["vector_blob"]),
-                        created_at=now,
-                    )
-                    for item in vectors
-                ]
-            )
+            vector_rows = [
+                {
+                    "id": str(uuid4()),
+                    "page_id": row.id,
+                    "knowledge_base_mapping_id": mapping_id,
+                    "section_index": int(item["section_index"]),
+                    "heading": item.get("heading"),
+                    "section_content": str(item["section_content"]),
+                    "content_hash": str(item["content_hash"]),
+                    "embedding_fingerprint": fingerprint,
+                    "embedding_dimensions": int(item["embedding_dimensions"]),
+                    "vector_blob": bytes(item["vector_blob"]),
+                    "created_at": now,
+                }
+                for item in vectors
+            ]
+            # Keep the page replacement atomic, but cap each SQL statement so
+            # large Wiki pages cannot flood the database with one huge insert.
+            for start in range(0, len(vector_rows), self.write_batch_size):
+                await session.execute(
+                    insert(LlmWikiWikiVectorRow),
+                    vector_rows[start : start + self.write_batch_size],
+                )
             state = await self._ensure_state(session, mapping_id)
             state.index_revision = int(state.index_revision or 0) + 1
             state.embedding_fingerprint = fingerprint