fix(avatar): cap knowledge files at 50MB

Reviewed-on: #14

digital-avatar-app/backend/database.py

@@ -1,16 +1,30 @@
 import os
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event
 from sqlalchemy.orm import sessionmaker, declarative_base, Session
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 DB_FILE = os.path.join(BASE_DIR, "avatar.db")
 DATABASE_URL = os.getenv("DATABASE_URL", f"sqlite:///{DB_FILE}")
 
+IS_SQLITE = DATABASE_URL.startswith("sqlite:")
 engine = create_engine(
     DATABASE_URL,
-    connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite:") else {},
+    connect_args={"check_same_thread": False, "timeout": 30} if IS_SQLITE else {},
 )
+
+
+if IS_SQLITE:
+    @event.listens_for(engine, "connect")
+    def _configure_sqlite_connection(dbapi_connection, _connection_record):
+        cursor = dbapi_connection.cursor()
+        try:
+            cursor.execute("PRAGMA synchronous=NORMAL")
+            cursor.execute("PRAGMA busy_timeout=30000")
+        finally:
+            cursor.close()
+
+
 SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
 Base = declarative_base()
 
@@ -26,6 +40,10 @@ def get_db():
 def init_db():
     import models
 
+    if IS_SQLITE:
+        with engine.connect() as conn:
+            conn.exec_driver_sql("PRAGMA journal_mode=WAL")
+            conn.commit()
     Base.metadata.create_all(bind=engine)
 
     # 轻量迁移：为已存在的表补充新列（SQLite 不支持自动 ALTER，逐列尝试）
@@ -35,6 +53,7 @@ def init_db():
         ("knowledge_docs", "embedding_model", "VARCHAR DEFAULT ''"),
         ("knowledge_docs", "chunk_count", "INTEGER DEFAULT 0"),
         ("knowledge_docs", "vectorized_at", "TIMESTAMP"),
+        ("knowledge_docs", "error_message", "VARCHAR DEFAULT ''"),
         ("avatars", "owner_id", "VARCHAR DEFAULT ''"),
         ("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
         ("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
@@ -45,6 +64,7 @@ def init_db():
         ("token_account", "total_consumed", "BIGINT DEFAULT 0"),
         ("token_account", "created_at", "TIMESTAMP"),
         ("token_account", "updated_at", "TIMESTAMP"),
+        ("takeover_messages", "attachment_id", "VARCHAR DEFAULT NULL"),
     )
     _normalize_optional_unique_values()
     _normalize_takeover_delays()

digital-avatar-app/backend/main.py

@@ -20,6 +20,7 @@ import routers.chat
 import routers.takeover
 from responses import ok
 from services.chat_attachment_service import purge_expired_chat_attachments
+from services.knowledge_vectorizer import knowledge_vectorizer
 from services.token_billing import DEFAULT_TOKEN_GRANT, release_stale_reservations
 
 logger = logging.getLogger(__name__)
@@ -131,6 +132,7 @@ def on_startup():
 
     init_db()
     seed()
+    knowledge_vectorizer.start()
 
     # Release stale resources when startup is invoked again by a reload/test.
     stop_takeover_scheduler()

digital-avatar-app/backend/models.py

@@ -120,6 +120,7 @@ class TakeoverMessage(Base):
     direction = Column(String, nullable=False)  # incoming | outgoing
     message_type = Column(Integer, default=0)
     content = Column(Text, default="")
+    attachment_id = Column(String, nullable=True)
     is_avatar = Column(Boolean, default=False)
     send_time = Column(DateTime, nullable=False)
     created_at = Column(DateTime, server_default=func.now())
@@ -189,6 +190,7 @@ class KnowledgeDoc(Base):
     file_size = Column(Integer, default=0)
     file_url = Column(String, default="")
     status = Column(String, default="uploaded")  # uploaded | parsing | ready | failed
+    error_message = Column(String, default="")  # 建立索引失败原因
     vectorized = Column(Boolean, default=False)  # 是否已向量化
     embedding_model = Column(String, default="")  # 向量模型标识
     chunk_count = Column(Integer, default=0)  # 切片数量
@@ -204,6 +206,7 @@ class KnowledgeDoc(Base):
             "fileSize": self.file_size,
             "fileUrl": self.file_url,
             "status": self.status,
+            "errorMessage": self.error_message or "",
             "vectorized": bool(self.vectorized),
             "embeddingModel": self.embedding_model,
             "chunkCount": self.chunk_count,
@@ -263,7 +266,7 @@ class ChatAttachment(Base):
     __tablename__ = "chat_attachments"
     id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex)
     avatar_id = Column(String, nullable=False, default="", index=True)
-    uploader_kind = Column(String, default="owner")  # owner | public
+    uploader_kind = Column(String, default="owner")  # owner | public | boxim
     filename = Column(String, default="")
     mime_type = Column(String, default="")
     file_size = Column(Integer, default=0)

digital-avatar-app/backend/routers/chat.py

@@ -47,6 +47,25 @@ QA_SEMANTIC_THRESHOLD = 0.72
 QA_MATCH_MARGIN = 0.06
 KNOWLEDGE_MIN_SCORE = float(os.getenv("KNOWLEDGE_MIN_SCORE", "0.42"))
 
+_IMAGE_ACCESS_DENIAL_PATTERNS = (
+    re.compile(
+        r"(?:我|目前|暂时|这里|本身|系统)?\s*(?:无法|不能|没法|不支持)\s*"
+        r"(?:直接)?\s*(?:查看|看到|看见|识别|读取|访问|打开|分析|理解)"
+        r"(?:\s*(?:或|、|/)\s*(?:查看|看到|看见|识别|读取|访问|打开|分析|理解))*\s*"
+        r"(?:你(?:发|提供|上传)的|这张|该|当前)?\s*(?:图片|图像|照片|影像|文件)"
+    ),
+    re.compile(
+        r"(?:我|这里|目前|暂时)?\s*(?:看不到|看不见|未看到|没有看到|没收到|未收到)\s*"
+        r"(?:你(?:发|提供|上传)的|这张|该|当前)?\s*(?:图片|图像|照片|影像)"
+    ),
+    re.compile(
+        r"\b(?:i\s+)?(?:can(?:not|'t)|am\s+unable\s+to)\s+(?:directly\s+)?"
+        r"(?:view|see|access|read|analy[sz]e|recogni[sz]e)\s+"
+        r"(?:the\s+|this\s+|your\s+)?(?:image|photo|picture|scan)\b",
+        re.IGNORECASE,
+    ),
+)
+
 _WRITING_SYSTEM_PATTERNS = {
     "han": re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]"),
     "latin": re.compile(r"[A-Za-z\u00c0-\u024f]"),
@@ -178,6 +197,75 @@ def _image_retrieval_question(question: str, image_contexts: list[dict]) -> str:
     return "\n".join(part for part in parts if part).strip()
 
 
+def _answer_denies_available_image(answer: str) -> bool:
+    """Reject only whole-image access denials, not uncertainty about one field."""
+    value = re.sub(r"\s+", " ", answer or "").strip()
+    return any(pattern.search(value) for pattern in _IMAGE_ACCESS_DENIAL_PATTERNS)
+
+
+def _compact_context_text(value: Any, limit: int) -> str:
+    lines = [re.sub(r"\s+", " ", line).strip() for line in str(value or "").splitlines()]
+    text = "\n".join(line for line in lines if line).strip()
+    return text[:limit].rstrip()
+
+
+def _grounded_image_fallback(question: str, image_contexts: list[dict]) -> str:
+    """Build a safe answer from completed vision data when the chat model contradicts it."""
+    summaries: list[str] = []
+    facts: list[str] = []
+    excerpts: list[str] = []
+    warnings: list[str] = []
+    for context in image_contexts:
+        summary = _compact_context_text(context.get("summary"), 500)
+        if summary:
+            summaries.append(summary)
+        structured = context.get("structuredData") or {}
+        if isinstance(structured, dict):
+            for fact in structured.get("key_facts") or []:
+                value = _compact_context_text(fact, 300)
+                if value:
+                    facts.append(value)
+        extracted = _compact_context_text(context.get("extractedText"), 900)
+        if extracted:
+            excerpts.append(extracted)
+        warning = _compact_context_text(context.get("warning"), 300)
+        if warning:
+            warnings.append(warning)
+
+    summaries = list(dict.fromkeys(summaries))
+    facts = list(dict.fromkeys(facts))[:6]
+    excerpts = list(dict.fromkeys(excerpts))
+    warnings = list(dict.fromkeys(warnings))
+    writing_system = _dominant_writing_system(question)
+
+    if writing_system == "latin":
+        parts = []
+        if summaries:
+            parts.append("From the image, I can confirm: " + " ".join(summaries))
+        if facts:
+            parts.append("Key details:\n" + "\n".join(
+                f"{index}. {fact}" for index, fact in enumerate(facts, 1)
+            ))
+        elif excerpts:
+            parts.append("Visible text:\n" + excerpts[0])
+        if warnings:
+            parts.append("Please note: " + " ".join(warnings))
+        return "\n".join(parts).strip() or "The image is available, but there is not enough clear detail to confirm more."
+
+    parts = []
+    if summaries:
+        parts.append("从这张图中可以确认：" + "；".join(summaries).rstrip("。；") + "。")
+    if facts:
+        parts.append("其中比较明确的信息有：\n" + "\n".join(
+            f"{index}. {fact}" for index, fact in enumerate(facts, 1)
+        ))
+    elif excerpts:
+        parts.append("图中可见的主要文字是：\n" + excerpts[0])
+    if warnings:
+        parts.append("需要注意：" + "；".join(warnings).rstrip("。；") + "。")
+    return "\n".join(parts).strip() or "这张图已经看到了，但目前能确认的清晰信息比较有限。"
+
+
 def _run_billed_vision_call(
     db: Session,
     avatar: Avatar,
@@ -236,11 +324,40 @@ async def _analyze_uploaded_image(
     max_bytes = max(1024, int(os.getenv("CHAT_IMAGE_MAX_BYTES", str(8 * 1024 * 1024))))
     content = await file.read(max_bytes + 1)
     filename = os.path.basename(file.filename or "图片")[:255]
+    try:
+        return _analyze_image_bytes(
+            db,
+            avatar,
+            content,
+            filename=filename,
+            mime_type=file.content_type or "",
+            uploader_kind=uploader_kind,
+        )
+    except ImageValidationError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except InsufficientTokensError:
+        raise
+    except RuntimeError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    finally:
+        content = b""
+
+
+def _analyze_image_bytes(
+    db: Session,
+    avatar: Avatar,
+    content: bytes,
+    *,
+    filename: str,
+    mime_type: str,
+    uploader_kind: str,
+) -> ChatAttachment:
+    """Analyze image bytes from either HTTP upload or BOXIM without persisting raw data."""
     attachment = ChatAttachment(
         avatar_id=avatar.id,
         uploader_kind=uploader_kind,
         filename=filename,
-        mime_type=(file.content_type or "")[:100],
+        mime_type=(mime_type or "")[:100],
         file_size=len(content),
         status="processing",
         expires_at=_attachment_expiry(),
@@ -312,7 +429,7 @@ async def _analyze_uploaded_image(
         attachment.status = "failed"
         attachment.warning = str(exc)
         db.commit()
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
+        raise
     except InsufficientTokensError:
         attachment.status = "failed"
         attachment.warning = "积分余额不足"
@@ -328,9 +445,7 @@ async def _analyze_uploaded_image(
             avatar.id,
             type(exc).__name__,
         )
-        raise HTTPException(status_code=502, detail=str(exc)) from exc
+        raise
-    finally:
-        content = b""
 
 
 def _normalize_question(value: str) -> str:
@@ -526,8 +641,10 @@ def _build_prompt(
     if image_contexts:
         image_material = json.dumps(image_contexts, ensure_ascii=False, default=str)
         system += (
-            "\n以下是当前会话图片经过视觉识别后得到的资料：\n"
+            "\n当前会话图片已经成功读取并完成内容识别，以下资料就是可直接使用的图片内容：\n"
             f"{image_material}"
+            "\n必须直接依据这些图片内容回答当前问题。禁止声称无法查看、看不到、未收到、无法识别、"
+            "无法读取或不能访问图片，也不要要求对方重新上传；只有资料明确标记读取失败时才可以请对方重发。"
             "\n图片资料可能包含 OCR 错字、模糊内容或用户尚未确认的信息，只能按可见内容谨慎表达。"
             "标准答题对中的事实优先级高于图片资料，知识库事实优先级高于模型推测；发生冲突时遵循更高优先级资料，"
             "并自然提醒对方核对原图。不得声称看到了图片中不存在的内容。"
@@ -788,6 +905,14 @@ def _resolve_reply(
         except Exception as exc:
             release_reservation(db, reservation, str(exc))
             raise
+    answer = str(answer or "").strip()
+    if image_contexts and _answer_denies_available_image(answer):
+        logger.warning(
+            "chat model contradicted ready image context avatar=%s source=%s",
+            avatar.id,
+            usage_source,
+        )
+        answer = _grounded_image_fallback(question, image_contexts)
     result = {
         "answer": answer,
         "source": "qa" if matched else (

digital-avatar-app/backend/routers/knowledge.py

@@ -1,8 +1,5 @@
 import os
-import json
-import logging
 import uuid
-from datetime import datetime, timezone
 
 from fastapi import APIRouter, UploadFile, File, Depends, Header, HTTPException
 from pydantic import BaseModel
@@ -12,16 +9,16 @@ from database import get_db
 from models import KnowledgeDoc, QAPair, KnowledgeChunk, Avatar, User
 from responses import ok, fail
 import embeddings
+from services.knowledge_vectorizer import knowledge_vectorizer
 
 router = APIRouter()
-logger = logging.getLogger(__name__)
-
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 UPLOAD_DIR = os.path.abspath(os.getenv("UPLOAD_DIR", os.path.join(BASE_DIR, "uploads")))
 os.makedirs(UPLOAD_DIR, exist_ok=True)
 
 ALLOWED_EXT = {".md", ".txt", ".pdf", ".doc", ".docx", ".xlsx"}
-MAX_UPLOAD_BYTES = 10 * 1024 * 1024
+MAX_UPLOAD_BYTES = 50 * 1024 * 1024
+UPLOAD_CHUNK_BYTES = 1024 * 1024
 
 
 class QAIn(BaseModel):
@@ -71,15 +68,6 @@ def list_docs(avatar_id: str, authorization: str = Header(None), db: Session = D
         .order_by(KnowledgeDoc.created_at.desc())
         .all()
     )
-    # Older synchronous uploads could be interrupted after persisting "parsing".
-    # New uploads are committed only after indexing finishes, so these rows are stale.
-    stale_docs = [doc for doc in docs if doc.status == "parsing"]
-    if stale_docs:
-        for doc in stale_docs:
-            doc.status = "failed"
-            doc.vectorized = False
-            doc.chunk_count = 0
-        db.commit()
     return ok([_doc_payload(d) for d in docs])
 
 
@@ -93,62 +81,62 @@ async def upload_doc(avatar_id: str, file: UploadFile = File(...), authorization
     os.makedirs(avatar_dir, exist_ok=True)
     stored = f"{uuid.uuid4().hex}{ext}"
     path = os.path.join(avatar_dir, stored)
-    content = await file.read()
+    file_size = 0
-    if len(content) > MAX_UPLOAD_BYTES:
+    try:
-        return fail("文件不能超过 10MB", code=400)
+        # Stream large files to disk so a 100MB upload does not occupy 100MB RAM.
         with open(path, "wb") as f:
-        f.write(content)
+            while chunk := await file.read(UPLOAD_CHUNK_BYTES):
+                file_size += len(chunk)
+                if file_size > MAX_UPLOAD_BYTES:
+                    raise ValueError("文件不能超过 50MB")
+                f.write(chunk)
+    except ValueError as exc:
+        if os.path.exists(path):
+            os.remove(path)
+        return fail(str(exc), code=400)
     doc = KnowledgeDoc(
         id=uuid.uuid4().hex,
         avatar_id=avatar_id,
         filename=file.filename,
         file_type=ext.lstrip("."),
-        file_size=len(content),
+        file_size=file_size,
         file_url=f"/api/files/{avatar_id}/{stored}",
         status="parsing",
     )
 
-    # Complete extraction and embedding before the first database commit so a
+    # Persist and acknowledge the upload first. Extraction and embeddings may take
-    # process restart cannot leave a permanent "parsing" row behind.
+    # minutes for a PDF and must never consume the browser request timeout.
-    try:
-        text = embeddings.extract_text(path, ext)
-        chunks = embeddings.chunk_text(text)
-        if not chunks:
-            raise ValueError("文档没有可建立索引的文字内容")
-        vectors = embeddings.embed(chunks)
-        if len(vectors) != len(chunks):
-            raise ValueError("向量服务返回数量与文档分段不一致")
-        doc.vectorized = True
-        doc.embedding_model = embeddings.MODEL
-        doc.chunk_count = len(chunks)
-        doc.vectorized_at = datetime.now(timezone.utc)
-        doc.status = "ready"
     db.add(doc)
-        for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
-            db.add(
-                KnowledgeChunk(
-                    doc_id=doc.id,
-                    avatar_id=avatar_id,
-                    content=chunk,
-                    vector=json.dumps(vector),
-                    chunk_index=i,
-                    embedding_model=embeddings.MODEL,
-                )
-            )
     db.commit()
     db.refresh(doc)
-    except Exception as exc:
+    knowledge_vectorizer.enqueue(doc.id)
-        db.rollback()
+
-        doc.status = "failed"
+    return ok(_doc_payload(doc))
+
+
+@router.post("/avatar/{avatar_id}/knowledge/docs/{doc_id}/retry")
+def retry_doc(avatar_id: str, doc_id: str, authorization: str = Header(None), db: Session = Depends(get_db)):
+    _require_owned_avatar(db, avatar_id, authorization)
+    doc = db.query(KnowledgeDoc).filter(
+        KnowledgeDoc.id == doc_id, KnowledgeDoc.avatar_id == avatar_id
+    ).first()
+    if not doc:
+        return fail("文档不存在", code=404)
+    if doc.vectorized and doc.status == "ready":
+        return ok(_doc_payload(doc))
+    stored_name = os.path.basename(doc.file_url or "")
+    if not stored_name or not os.path.isfile(os.path.join(UPLOAD_DIR, avatar_id, stored_name)):
+        return fail("原文件不可用，请重新上传", code=400)
+    db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == doc.id).delete()
+    doc.status = "parsing"
     doc.vectorized = False
     doc.embedding_model = ""
     doc.chunk_count = 0
     doc.vectorized_at = None
-        db.add(doc)
+    doc.error_message = ""
     db.commit()
     db.refresh(doc)
-        logger.exception("knowledge vectorization failed for %s: %s", doc.id, exc)
+    knowledge_vectorizer.enqueue(doc.id)
-
     return ok(_doc_payload(doc))
 
 

digital-avatar-app/backend/services/boxim_image_service.py

@@ -0,0 +1,151 @@
+"""Parse and safely download image payloads from BOXIM private messages."""
+
+import ipaddress
+import json
+import os
+import socket
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+from urllib.parse import unquote, urljoin, urlsplit
+
+import httpx
+
+
+MAX_REDIRECTS = 3
+
+
+class BoxIMImageError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class DownloadedBoxIMImage:
+    content: bytes
+    filename: str
+    mime_type: str
+    source_url: str
+
+
+def parse_boxim_image_url(content: str, *, base_url: str = "") -> str:
+    try:
+        payload = json.loads(content or "")
+    except (TypeError, ValueError) as exc:
+        raise BoxIMImageError("BOXIM 图片消息格式无效") from exc
+    if not isinstance(payload, dict):
+        raise BoxIMImageError("BOXIM 图片消息格式无效")
+
+    value = payload.get("originUrl") or payload.get("thumbUrl") or payload.get("url")
+    if not isinstance(value, str) or not value.strip():
+        raise BoxIMImageError("BOXIM 图片消息缺少图片地址")
+    value = value.strip()
+    if value.startswith("/"):
+        if not base_url:
+            raise BoxIMImageError("BOXIM 图片地址不完整")
+        value = urljoin(f"{base_url.rstrip('/')}/", value)
+    return value
+
+
+def _configured_hosts(name: str) -> set[str]:
+    return {
+        value.strip().lower().rstrip(".")
+        for value in os.getenv(name, "").split(",")
+        if value.strip()
+    }
+
+
+def _host_matches(host: str, configured: set[str]) -> bool:
+    return any(host == value or host.endswith(f".{value}") for value in configured)
+
+
+def _resolved_addresses(host: str, port: int) -> set[ipaddress.IPv4Address | ipaddress.IPv6Address]:
+    try:
+        return {
+            ipaddress.ip_address(item[4][0])
+            for item in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+        }
+    except (OSError, ValueError) as exc:
+        raise BoxIMImageError("BOXIM 图片地址无法解析") from exc
+
+
+def _is_safe_remote_url(url: str) -> None:
+    parsed = urlsplit(url)
+    scheme = parsed.scheme.lower()
+    allow_http = os.getenv("BOXIM_IMAGE_ALLOW_HTTP", "").lower() in {"1", "true", "yes"}
+    if scheme not in ({"https", "http"} if allow_http else {"https"}):
+        raise BoxIMImageError("BOXIM 图片地址必须使用 HTTPS")
+    if parsed.username or parsed.password or not parsed.hostname:
+        raise BoxIMImageError("BOXIM 图片地址无效")
+
+    host = parsed.hostname.lower().rstrip(".")
+    allowed_hosts = _configured_hosts("BOXIM_IMAGE_ALLOWED_HOSTS")
+    if allowed_hosts and not _host_matches(host, allowed_hosts):
+        raise BoxIMImageError("BOXIM 图片地址不在允许的域名范围内")
+
+    private_hosts = _configured_hosts("BOXIM_IMAGE_PRIVATE_HOSTS")
+    try:
+        addresses = {ipaddress.ip_address(host)}
+    except ValueError:
+        addresses = _resolved_addresses(host, parsed.port or (443 if scheme == "https" else 80))
+    if not addresses:
+        raise BoxIMImageError("BOXIM 图片地址无法解析")
+    if _host_matches(host, private_hosts):
+        return
+    if any(not address.is_global for address in addresses):
+        raise BoxIMImageError("BOXIM 图片地址指向受限网络")
+
+
+def _filename_from_url(url: str) -> str:
+    value = unquote(PurePosixPath(urlsplit(url).path).name).strip()
+    value = value.replace("\x00", "")
+    return (value or "boxim-image")[:255]
+
+
+def download_boxim_image(
+    content: str,
+    *,
+    base_url: str = "",
+    transport: httpx.BaseTransport | None = None,
+) -> DownloadedBoxIMImage:
+    """Download one BOXIM image without redirects or oversized responses escaping checks."""
+    url = parse_boxim_image_url(content, base_url=base_url)
+    max_bytes = max(1024, int(os.getenv("CHAT_IMAGE_MAX_BYTES", str(8 * 1024 * 1024))))
+    timeout = max(1.0, min(float(os.getenv("BOXIM_IMAGE_TIMEOUT_SECONDS", "15")), 60.0))
+
+    with httpx.Client(
+        timeout=timeout,
+        follow_redirects=False,
+        trust_env=False,
+        transport=transport,
+    ) as client:
+        for _ in range(MAX_REDIRECTS + 1):
+            _is_safe_remote_url(url)
+            try:
+                with client.stream("GET", url, headers={"Accept": "image/*"}) as response:
+                    if response.status_code in {301, 302, 303, 307, 308}:
+                        location = response.headers.get("location", "").strip()
+                        if not location:
+                            raise BoxIMImageError("BOXIM 图片跳转地址无效")
+                        url = urljoin(url, location)
+                        continue
+                    response.raise_for_status()
+                    raw_length = response.headers.get("content-length", "")
+                    if raw_length.isdigit() and int(raw_length) > max_bytes:
+                        raise BoxIMImageError("BOXIM 图片超过大小限制")
+                    chunks = bytearray()
+                    for chunk in response.iter_bytes():
+                        chunks.extend(chunk)
+                        if len(chunks) > max_bytes:
+                            raise BoxIMImageError("BOXIM 图片超过大小限制")
+                    if not chunks:
+                        raise BoxIMImageError("BOXIM 图片内容为空")
+                    return DownloadedBoxIMImage(
+                        content=bytes(chunks),
+                        filename=_filename_from_url(url),
+                        mime_type=response.headers.get("content-type", "").split(";", 1)[0][:100],
+                        source_url=url,
+                    )
+            except BoxIMImageError:
+                raise
+            except (httpx.HTTPError, OSError) as exc:
+                raise BoxIMImageError("BOXIM 图片下载失败") from exc
+    raise BoxIMImageError("BOXIM 图片跳转次数过多")

digital-avatar-app/backend/services/knowledge_vectorizer.py

@@ -0,0 +1,125 @@
+"""Durable, serial knowledge-document indexing for the avatar knowledge base."""
+
+import json
+import logging
+import os
+import queue
+import threading
+from datetime import datetime, timezone
+
+from database import SessionLocal
+from models import KnowledgeChunk, KnowledgeDoc
+import embeddings
+
+logger = logging.getLogger(__name__)
+
+BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+UPLOAD_DIR = os.path.abspath(
+    os.getenv("UPLOAD_DIR", os.path.join(BACKEND_DIR, "routers", "uploads"))
+)
+
+
+class KnowledgeVectorizer:
+    """Indexes one document at a time so slow providers cannot block uploads."""
+
+    def __init__(self):
+        self._queue: queue.Queue[str] = queue.Queue()
+        self._queued: set[str] = set()
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+
+    def start(self):
+        if self._thread and self._thread.is_alive():
+            return
+        self._thread = threading.Thread(
+            target=self._run, name="knowledge-vectorizer", daemon=True
+        )
+        self._thread.start()
+        db = SessionLocal()
+        try:
+            # A process restart must not abandon documents already accepted by upload.
+            for (doc_id,) in db.query(KnowledgeDoc.id).filter(KnowledgeDoc.status == "parsing"):
+                self.enqueue(doc_id)
+        finally:
+            db.close()
+
+    def enqueue(self, doc_id: str):
+        with self._lock:
+            if doc_id in self._queued:
+                return
+            self._queued.add(doc_id)
+        self._queue.put(doc_id)
+
+    def _run(self):
+        while True:
+            doc_id = self._queue.get()
+            try:
+                self.vectorize_document(doc_id)
+            except Exception:
+                logger.exception("Unexpected knowledge vectorizer failure for %s", doc_id)
+            finally:
+                with self._lock:
+                    self._queued.discard(doc_id)
+                self._queue.task_done()
+
+    def vectorize_document(self, doc_id: str):
+        db = SessionLocal()
+        try:
+            doc = db.get(KnowledgeDoc, doc_id)
+            if not doc or doc.status != "parsing":
+                return
+
+            stored_name = os.path.basename(doc.file_url or "")
+            path = os.path.join(UPLOAD_DIR, doc.avatar_id, stored_name)
+            if not stored_name or not os.path.isfile(path):
+                raise FileNotFoundError("原文件不可用，请重新上传")
+
+            text = embeddings.extract_text(path, f".{doc.file_type}")
+            chunks = embeddings.chunk_text(text)
+            if not chunks:
+                raise ValueError("文档没有可建立索引的文字内容")
+            vectors = embeddings.embed(chunks)
+            if len(vectors) != len(chunks):
+                raise ValueError("向量服务返回数量与文档分段不一致")
+
+            # Commit the document and every chunk together. Chat only sees complete indexes.
+            db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == doc.id).delete()
+            db.add_all(
+                [
+                    KnowledgeChunk(
+                        doc_id=doc.id,
+                        avatar_id=doc.avatar_id,
+                        content=chunk,
+                        vector=json.dumps(vector),
+                        chunk_index=index,
+                        embedding_model=embeddings.MODEL,
+                    )
+                    for index, (chunk, vector) in enumerate(zip(chunks, vectors))
+                ]
+            )
+            doc.vectorized = True
+            doc.embedding_model = embeddings.MODEL
+            doc.chunk_count = len(chunks)
+            doc.vectorized_at = datetime.now(timezone.utc)
+            doc.status = "ready"
+            doc.error_message = ""
+            db.commit()
+            logger.info("Knowledge document %s indexed with %s chunks", doc.id, len(chunks))
+        except Exception as exc:
+            db.rollback()
+            failed_doc = db.get(KnowledgeDoc, doc_id)
+            if failed_doc:
+                db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == failed_doc.id).delete()
+                failed_doc.status = "failed"
+                failed_doc.vectorized = False
+                failed_doc.embedding_model = ""
+                failed_doc.chunk_count = 0
+                failed_doc.vectorized_at = None
+                failed_doc.error_message = str(exc)[:300] or "建立知识索引失败"
+                db.commit()
+            logger.exception("Knowledge vectorization failed for %s: %s", doc_id, exc)
+        finally:
+            db.close()
+
+
+knowledge_vectorizer = KnowledgeVectorizer()

digital-avatar-app/backend/services/takeover_service.py

@@ -3,6 +3,7 @@
 import asyncio
 import hashlib
 import logging
+import os
 import re
 import secrets
 import time
@@ -13,12 +14,19 @@ from sqlalchemy.orm import Session
 
 from models import (
     Avatar,
+    ChatAttachment,
     TakeoverCursor,
     TakeoverMessage,
     TakeoverReplyTask,
     User,
 )
 from services.boxim_client import BoxIMClient, BoxIMError
+from services.boxim_image_service import (
+    BoxIMImageError,
+    download_boxim_image,
+    parse_boxim_image_url,
+)
+from services.vision_service import ImageValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -37,6 +45,19 @@ HUMAN_PAUSE_SECONDS = 600
 RATE_LIMIT_WINDOW_SECONDS = 300
 RATE_LIMIT_MAX_REPLIES = 5
 AVATAR_LOCAL_ID_PREFIX = "880"
+BOXIM_TEXT_MESSAGE_TYPE = 0
+BOXIM_IMAGE_MESSAGE_TYPE = 1
+BOXIM_IMAGE_PROMPT = "请看看这张图片。"
+BOXIM_IMAGE_UNAVAILABLE_REPLY = "这张图片我暂时没看清，麻烦重新发送一张清晰的原图。"
+IMAGE_CONTEXT_LOOKBACK_SECONDS = 1800
+IMAGE_REFERENCE_LOOKBACK_SECONDS = 172_800
+MAX_RECENT_IMAGE_CONTEXTS = 3
+
+_IMAGE_REFERENCE_PATTERN = re.compile(
+    r"(?:图片|图像|照片|截图|这张图|刚才.{0,8}图|病例|病历|检查单|检验单|化验单|报告|影像|"
+    r"\b(?:image|photo|picture|screenshot|scan|report)\b)",
+    re.IGNORECASE,
+)
 
 
 def _utcnow() -> datetime:
@@ -107,6 +128,18 @@ def _configured_reply_delay(avatar: Avatar, fallback: int | None = None) -> int:
     return delay
 
 
+def _event_prompt(event: TakeoverMessage) -> str:
+    if event.message_type == BOXIM_TEXT_MESSAGE_TYPE:
+        return event.content.strip()
+    if event.message_type == BOXIM_IMAGE_MESSAGE_TYPE:
+        return BOXIM_IMAGE_PROMPT
+    return ""
+
+
+def _references_recent_image(value: str) -> bool:
+    return bool(_IMAGE_REFERENCE_PATTERN.search(value or ""))
+
+
 class TakeoverService:
     """Poll BOXIM, honor the owner grace period, then generate and send one reply."""
 
@@ -129,6 +162,7 @@ class TakeoverService:
         self._sessions: dict[str, dict] = {}
         self._poll_lock = asyncio.Lock()
         self._process_lock = asyncio.Lock()
+        self._persist_lock = asyncio.Lock()
 
     async def poll_and_process_messages(self):
         """Run one complete cycle for callers that do not use the split scheduler."""
@@ -389,13 +423,6 @@ class TakeoverService:
             max_message_id = _numeric_id(cursor.last_message_id)
             read_receipts: dict[str, int] = {}
             for message in messages:
-                self._record_message(
-                    db,
-                    avatar,
-                    cursor.boxim_owner_id,
-                    message,
-                    schedule_reply=not priming,
-                )
                 message_id = _numeric_id(message.get("id"))
                 max_message_id = max(max_message_id, message_id)
                 send_id = str(message.get("sendId") or "")
@@ -410,6 +437,17 @@ class TakeoverService:
                     session["access_token"], peer_id, message_id
                 )
 
+            # Keep SQLite write transactions short. The read-receipt request above
+            # can block on the network and must not hold the database write lock.
+            async with self._persist_lock:
+                for message in messages:
+                    self._record_message(
+                        db,
+                        avatar,
+                        cursor.boxim_owner_id,
+                        message,
+                        schedule_reply=not priming,
+                    )
                 cursor.last_message_id = str(max_message_id)
                 cursor.initialized = True
                 cursor.last_polled_at = self.now()
@@ -498,7 +536,26 @@ class TakeoverService:
             if not is_avatar:
                 self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_replied")
             return
-        if not schedule_reply or event.message_type != 0 or not event.content.strip():
+        if not schedule_reply or event.message_type not in {
+            BOXIM_TEXT_MESSAGE_TYPE,
+            BOXIM_IMAGE_MESSAGE_TYPE,
+        }:
+            return
+        if event.message_type == BOXIM_TEXT_MESSAGE_TYPE and not event.content.strip():
+            return
+        if event.message_type == BOXIM_IMAGE_MESSAGE_TYPE:
+            try:
+                parse_boxim_image_url(
+                    event.content,
+                    base_url=getattr(self.boxim, "im_base_url", ""),
+                )
+            except BoxIMImageError as exc:
+                logger.warning(
+                    "Ignored invalid BOXIM image message %s for avatar %s: %s",
+                    message_id,
+                    avatar.id,
+                    exc,
+                )
                 return
         if (now - send_time).total_seconds() > self.max_message_age_seconds:
             logger.info(
@@ -606,7 +663,16 @@ class TakeoverService:
                 task.status = "cancelled"
                 task.cancel_reason = "newer_incoming_message"
                 task.locked_at = None
-        prompt_parts.append(event.content.strip())
+        if event.message_type == BOXIM_TEXT_MESSAGE_TYPE:
+            for image_event in self._recent_unhandled_images(
+                db,
+                avatar,
+                event,
+                source_ids,
+            ):
+                prompt_parts.append(_event_prompt(image_event))
+                source_ids.append(image_event.boxim_message_id)
+        prompt_parts.append(_event_prompt(event))
         source_ids.append(event.boxim_message_id)
         prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:]
         due_at = max(
@@ -631,6 +697,68 @@ class TakeoverService:
             )
         )
 
+    @staticmethod
+    def _recent_unhandled_images(
+        db: Session,
+        avatar: Avatar,
+        event: TakeoverMessage,
+        current_source_ids: list[str],
+    ) -> list[TakeoverMessage]:
+        """Recover missed images, or reuse a referenced image from the last two days."""
+        references_image = _references_recent_image(event.content)
+        lookback_seconds = (
+            IMAGE_REFERENCE_LOOKBACK_SECONDS
+            if references_image
+            else IMAGE_CONTEXT_LOOKBACK_SECONDS
+        )
+        threshold = event.send_time - timedelta(seconds=lookback_seconds)
+        candidates = (
+            db.query(TakeoverMessage)
+            .filter(
+                TakeoverMessage.avatar_id == avatar.id,
+                TakeoverMessage.owner_id == avatar.owner_id,
+                TakeoverMessage.peer_id == event.peer_id,
+                TakeoverMessage.direction == "incoming",
+                TakeoverMessage.message_type == BOXIM_IMAGE_MESSAGE_TYPE,
+                TakeoverMessage.is_avatar.is_(False),
+                TakeoverMessage.send_time >= threshold,
+                TakeoverMessage.send_time <= event.send_time,
+            )
+            .order_by(TakeoverMessage.send_time.desc())
+            .limit(MAX_RECENT_IMAGE_CONTEXTS)
+            .all()
+        )
+        if not candidates:
+            return []
+
+        current_ids = set(current_source_ids)
+        if references_image:
+            return [
+                image
+                for image in reversed(candidates)
+                if image.boxim_message_id not in current_ids
+            ]
+
+        handled_ids = set(current_ids)
+        task_sources = (
+            db.query(TakeoverReplyTask.source_message_ids)
+            .filter(
+                TakeoverReplyTask.avatar_id == avatar.id,
+                TakeoverReplyTask.owner_id == avatar.owner_id,
+                TakeoverReplyTask.peer_id == event.peer_id,
+                TakeoverReplyTask.created_at >= threshold,
+            )
+            .all()
+        )
+        for (source_message_ids,) in task_sources:
+            handled_ids.update(source_message_ids or [])
+
+        return [
+            image
+            for image in reversed(candidates)
+            if image.boxim_message_id not in handled_ids
+        ]
+
     async def _prepare_replies(self) -> int:
         db = self.session_factory()
         try:
@@ -665,6 +793,50 @@ class TakeoverService:
         results = await asyncio.gather(*(generate(task_id) for task_id in task_ids))
         return sum(bool(result) for result in results)
 
+    def _takeover_image_attachment(
+        self,
+        db: Session,
+        avatar: Avatar,
+        event: TakeoverMessage,
+    ) -> ChatAttachment:
+        now = self.now()
+        if event.attachment_id:
+            cached = db.get(ChatAttachment, event.attachment_id)
+            if cached and cached.status == "ready" and cached.expires_at > now:
+                cached.used_at = now
+                db.commit()
+                return cached
+
+        downloaded = download_boxim_image(
+            event.content,
+            base_url=getattr(
+                self.boxim,
+                "im_base_url",
+                os.getenv("BOXIM_API_BASE_URL", "https://im.99hui.com/api"),
+            ),
+        )
+        from routers.chat import _analyze_image_bytes
+
+        attachment = _analyze_image_bytes(
+            db,
+            avatar,
+            downloaded.content,
+            filename=downloaded.filename,
+            mime_type=downloaded.mime_type,
+            uploader_kind="boxim",
+        )
+        event.attachment_id = attachment.id
+        attachment.used_at = now
+        db.commit()
+        logger.info(
+            "BOXIM image analyzed message=%s attachment=%s avatar=%s category=%s",
+            event.boxim_message_id,
+            attachment.id,
+            avatar.id,
+            attachment.category,
+        )
+        return attachment
+
     def _generate_reply(self, task_id: str) -> bool:
         db = self.session_factory()
         try:
@@ -683,6 +855,21 @@ class TakeoverService:
             db.commit()
 
             excluded_ids = set(task.source_message_ids or [])
+            source_events = {
+                event.boxim_message_id: event
+                for event in (
+                    db.query(TakeoverMessage)
+                    .filter(
+                        TakeoverMessage.owner_id == task.owner_id,
+                        TakeoverMessage.peer_id == task.peer_id,
+                        TakeoverMessage.avatar_id == task.avatar_id,
+                        TakeoverMessage.boxim_message_id.in_(excluded_ids),
+                    )
+                    .all()
+                    if excluded_ids
+                    else []
+                )
+            }
             events = (
                 db.query(TakeoverMessage)
                 .filter(
@@ -694,9 +881,31 @@ class TakeoverService:
                 .limit(30)
                 .all()
             )
+            image_attachments = []
+            image_failed = False
+            for message_id in (task.source_message_ids or [])[-3:]:
+                event = source_events.get(message_id)
+                if not event or event.message_type != BOXIM_IMAGE_MESSAGE_TYPE:
+                    continue
+                try:
+                    image_attachments.append(
+                        self._takeover_image_attachment(db, avatar, event)
+                    )
+                except (BoxIMImageError, ImageValidationError) as exc:
+                    image_failed = True
+                    logger.warning(
+                        "BOXIM image unavailable message=%s avatar=%s: %s",
+                        event.boxim_message_id,
+                        avatar.id,
+                        exc,
+                    )
             history = []
             for event in reversed(events):
-                if event.boxim_message_id in excluded_ids or not event.content.strip():
+                if (
+                    event.boxim_message_id in excluded_ids
+                    or event.message_type != BOXIM_TEXT_MESSAGE_TYPE
+                    or not event.content.strip()
+                ):
                     continue
                 if event.direction == "incoming" and event.is_avatar:
                     continue
@@ -708,9 +917,20 @@ class TakeoverService:
                 )
             history = history[-10:]
 
-            from routers.chat import _resolve_reply
+            from routers.chat import _attachment_contexts, _resolve_reply
 
-            result = _resolve_reply(db, avatar, task.prompt, history, usage_source="takeover")
+            image_contexts = _attachment_contexts(image_attachments)
+            if image_failed and not image_contexts:
+                answer = BOXIM_IMAGE_UNAVAILABLE_REPLY
+            else:
+                result = _resolve_reply(
+                    db,
+                    avatar,
+                    task.prompt,
+                    history,
+                    usage_source="takeover",
+                    image_contexts=image_contexts,
+                )
                 answer = _plain_text_reply(result.get("answer", ""))
             db.refresh(task)
             if task.status != "generating":

digital-avatar-app/backend/tests/test_boxim_image_service.py

@@ -0,0 +1,71 @@
+import ipaddress
+import json
+
+import httpx
+import pytest
+
+from services.boxim_image_service import (
+    BoxIMImageError,
+    download_boxim_image,
+    parse_boxim_image_url,
+)
+
+
+def test_parse_boxim_image_prefers_origin_and_supports_relative_url():
+    content = json.dumps({"originUrl": "/files/original.png", "thumbUrl": "/thumb.png"})
+    assert parse_boxim_image_url(content, base_url="https://im.example/api") == (
+        "https://im.example/files/original.png"
+    )
+
+
+def test_download_boxim_image_streams_public_https(monkeypatch):
+    monkeypatch.setattr(
+        "services.boxim_image_service._resolved_addresses",
+        lambda _host, _port: {ipaddress.ip_address("8.8.8.8")},
+    )
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(
+            200,
+            headers={"content-type": "image/png"},
+            content=b"png-bytes",
+            request=request,
+        )
+    )
+
+    image = download_boxim_image(
+        json.dumps({"originUrl": "https://cdn.example/case%20photo.png"}),
+        transport=transport,
+    )
+
+    assert image.content == b"png-bytes"
+    assert image.filename == "case photo.png"
+    assert image.mime_type == "image/png"
+
+
+def test_download_boxim_image_rejects_private_network_url():
+    with pytest.raises(BoxIMImageError, match="受限网络"):
+        download_boxim_image(
+            json.dumps({"originUrl": "https://127.0.0.1/private.png"}),
+            transport=httpx.MockTransport(lambda request: httpx.Response(200, request=request)),
+        )
+
+
+def test_download_boxim_image_stops_oversized_stream(monkeypatch):
+    monkeypatch.setenv("CHAT_IMAGE_MAX_BYTES", "1024")
+    monkeypatch.setattr(
+        "services.boxim_image_service._resolved_addresses",
+        lambda _host, _port: {ipaddress.ip_address("8.8.8.8")},
+    )
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(
+            200,
+            headers={"content-length": "2048"},
+            request=request,
+        )
+    )
+
+    with pytest.raises(BoxIMImageError, match="超过大小限制"):
+        download_boxim_image(
+            json.dumps({"originUrl": "https://cdn.example/large.png"}),
+            transport=transport,
+        )

digital-avatar-app/backend/tests/test_chat_images.py

@@ -12,11 +12,13 @@ from main import app
 from models import ChatAttachment
 from routers.chat import (
     ChatIn,
+    _answer_denies_available_image,
     _attachment_contexts,
     _load_chat_attachments,
     _resolve_reply,
 )
 from services.chat_attachment_service import purge_expired_chat_attachments
+from services.token_billing import InsufficientTokensError
 from services.vision_service import PreparedImage
 
 
@@ -75,6 +77,22 @@ def test_non_owner_cannot_upload_chat_image(authorization_context):
     assert response.status_code == 403
 
 
+def test_image_upload_preserves_insufficient_points_response(authorization_context):
+    context = authorization_context
+    with patch(
+        "routers.chat._analyze_image_bytes",
+        side_effect=InsufficientTokensError("积分余额不足"),
+    ):
+        response = client.post(
+            f"/api/avatar/{context['avatar'].id}/chat/images",
+            headers=context["owner_headers"],
+            files={"file": ("private.png", b"image-bytes", "image/png")},
+        )
+
+    assert response.status_code == 402
+    assert response.json()["detail"] == "积分余额不足"
+
+
 def test_public_share_can_upload_without_exposing_analysis_details(authorization_context):
     context = authorization_context
     db = SessionLocal()
@@ -257,6 +275,47 @@ def test_image_context_keeps_standard_answer_authoritative():
     assert "标准答题对中的事实优先级高于图片资料" in system
 
 
+def test_ready_image_context_never_returns_whole_image_access_denial():
+    avatar = SimpleNamespace(
+        id="avatar-vision",
+        name="测试分身",
+        description="产品顾问",
+        config={},
+    )
+    model = Mock(return_value="抱歉，我无法查看或识别图片，请重新上传。")
+    result = _resolve_reply(
+        None,
+        avatar,
+        "请看看这张图片",
+        [],
+        qa_pairs=[],
+        search_fn=Mock(return_value=[]),
+        model_client=model,
+        image_contexts=[{
+            "id": "attachment",
+            "filename": "report.jpg",
+            "category": "medical_document",
+            "summary": "一份耳鼻喉科门诊记录",
+            "extractedText": "主诉：咽痛三天",
+            "structuredData": {"key_facts": ["主诉为咽痛三天"]},
+            "warning": "请核对原始资料",
+        }],
+    )
+
+    assert result["source"] == "vision"
+    assert "一份耳鼻喉科门诊记录" in result["answer"]
+    assert "主诉为咽痛三天" in result["answer"]
+    assert "无法查看" not in result["answer"]
+    system = model.call_args.kwargs["messages"][0]["content"]
+    assert "当前会话图片已经成功读取" in system
+    assert "禁止声称无法查看" in system
+
+
+def test_image_denial_detector_allows_uncertain_field_in_ready_image():
+    assert _answer_denies_available_image("我无法查看这张图片") is True
+    assert _answer_denies_available_image("图片中患者姓名无法辨认，主诉为咽痛三天。") is False
+
+
 def test_attachment_context_does_not_expose_internal_fields():
     row = SimpleNamespace(
         id="attachment",

digital-avatar-app/backend/tests/test_knowledge_storage.py

@@ -8,6 +8,7 @@ from database import SessionLocal
 from main import app
 from models import Avatar, KnowledgeChunk, KnowledgeDoc, QAPair
 from routers.knowledge import _doc_payload
+from services.knowledge_vectorizer import knowledge_vectorizer
 
 
 client = TestClient(app)
@@ -31,14 +32,14 @@ def test_doc_payload_reports_whether_the_persisted_file_exists(tmp_path: Path):
         assert _doc_payload(doc)["filePresent"] is True
 
 
-def test_upload_marks_vectorization_failure_instead_of_staying_processing(
+def test_upload_returns_before_background_vectorization(
     tmp_path: Path,
     authorization_context,
 ):
     context = authorization_context
     with (
         patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
-        patch("routers.knowledge.embeddings.embed", side_effect=RuntimeError("provider unavailable")),
+        patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
     ):
         response = client.post(
             f"/api/avatar/{context['avatar'].id}/knowledge/docs",
@@ -47,14 +48,15 @@ def test_upload_marks_vectorization_failure_instead_of_staying_processing(
         )
 
     payload = response.json()["data"]
-    assert payload["status"] == "failed"
+    assert payload["status"] == "parsing"
     assert payload["vectorized"] is False
     assert payload["chunkCount"] == 0
+    enqueue.assert_called_once_with(payload["id"])
 
     db = SessionLocal()
     try:
         stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
-        assert stored.status == "failed"
+        assert stored.status == "parsing"
         assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 0
         db.delete(stored)
         db.commit()
@@ -62,14 +64,37 @@ def test_upload_marks_vectorization_failure_instead_of_staying_processing(
         db.close()
 
 
-def test_markdown_upload_commits_ready_document_and_chunks_together(
+def test_upload_rejects_oversize_file_before_queuing_indexing(
     tmp_path: Path,
     authorization_context,
 ):
     context = authorization_context
     with (
         patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
-        patch("routers.knowledge.embeddings.embed", return_value=[[1.0, 0.0]]),
+        patch("routers.knowledge.MAX_UPLOAD_BYTES", 4),
+        patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
+    ):
+        response = client.post(
+            f"/api/avatar/{context['avatar'].id}/knowledge/docs",
+            headers=context["owner_headers"],
+            files={"file": ("oversize.md", b"12345", "text/markdown")},
+        )
+
+    payload = response.json()
+    assert payload["code"] == 400
+    assert payload["message"] == "文件不能超过 50MB"
+    enqueue.assert_not_called()
+    assert not list((tmp_path / context["avatar"].id).glob("*"))
+
+
+def test_background_vectorizer_commits_ready_document_and_chunks_together(
+    tmp_path: Path,
+    authorization_context,
+):
+    context = authorization_context
+    with (
+        patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
+        patch("routers.knowledge.knowledge_vectorizer.enqueue"),
     ):
         response = client.post(
             f"/api/avatar/{context['avatar'].id}/knowledge/docs",
@@ -78,14 +103,19 @@ def test_markdown_upload_commits_ready_document_and_chunks_together(
         )
 
     payload = response.json()["data"]
-    assert payload["status"] == "ready"
+    assert payload["status"] == "parsing"
-    assert payload["vectorized"] is True
+    with (
-    assert payload["chunkCount"] == 1
+        patch("services.knowledge_vectorizer.UPLOAD_DIR", str(tmp_path)),
+        patch("services.knowledge_vectorizer.embeddings.embed", return_value=[[1.0, 0.0]]),
+    ):
+        knowledge_vectorizer.vectorize_document(payload["id"])
 
     db = SessionLocal()
     try:
         stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
         assert stored.status == "ready"
+        assert stored.vectorized is True
+        assert stored.chunk_count == 1
         assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 1
         db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).delete()
         db.delete(stored)
@@ -94,6 +124,87 @@ def test_markdown_upload_commits_ready_document_and_chunks_together(
         db.close()
 
 
+def test_background_vectorizer_keeps_failure_reason_for_retry(
+    tmp_path: Path,
+    authorization_context,
+):
+    context = authorization_context
+    with (
+        patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
+        patch("routers.knowledge.knowledge_vectorizer.enqueue"),
+    ):
+        response = client.post(
+            f"/api/avatar/{context['avatar'].id}/knowledge/docs",
+            headers=context["owner_headers"],
+            files={"file": ("knowledge.md", b"# Knowledge\n\nTest content", "text/markdown")},
+        )
+
+    payload = response.json()["data"]
+    with (
+        patch("services.knowledge_vectorizer.UPLOAD_DIR", str(tmp_path)),
+        patch("services.knowledge_vectorizer.embeddings.embed", side_effect=RuntimeError("provider unavailable")),
+    ):
+        knowledge_vectorizer.vectorize_document(payload["id"])
+
+    db = SessionLocal()
+    try:
+        stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
+        assert stored.status == "failed"
+        assert stored.error_message == "provider unavailable"
+        assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 0
+        db.delete(stored)
+        db.commit()
+    finally:
+        db.close()
+
+
+def test_retry_queues_a_failed_document_again(
+    tmp_path: Path,
+    authorization_context,
+):
+    context = authorization_context
+    document_id = f"retry-doc-{context['suffix']}"
+    avatar_dir = tmp_path / context["avatar"].id
+    avatar_dir.mkdir()
+    (avatar_dir / "retry.md").write_text("retry content", encoding="utf-8")
+    db = SessionLocal()
+    try:
+        db.add(
+            KnowledgeDoc(
+                id=document_id,
+                avatar_id=context["avatar"].id,
+                filename="retry.md",
+                file_type="md",
+                file_url=f"/api/files/{context['avatar'].id}/retry.md",
+                status="failed",
+                error_message="provider unavailable",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    with (
+        patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
+        patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
+    ):
+        response = client.post(
+            f"/api/avatar/{context['avatar'].id}/knowledge/docs/{document_id}/retry",
+            headers=context["owner_headers"],
+        )
+
+    payload = response.json()["data"]
+    assert payload["status"] == "parsing"
+    assert payload["errorMessage"] == ""
+    enqueue.assert_called_once_with(document_id)
+    db = SessionLocal()
+    try:
+        db.query(KnowledgeDoc).filter(KnowledgeDoc.id == document_id).delete()
+        db.commit()
+    finally:
+        db.close()
+
+
 def test_each_avatar_has_an_independent_document_and_qa_scope(authorization_context):
     context = authorization_context
     first_avatar_id = context["avatar"].id

digital-avatar-app/backend/tests/test_takeover_service.py

@@ -1,6 +1,7 @@
 """End-to-end service tests for BOXIM takeover timing and human priority."""
 
 import asyncio
+import json
 from datetime import datetime, timedelta, timezone
 from threading import Barrier
 from unittest.mock import AsyncMock, patch
@@ -10,8 +11,9 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
 from database import Base
-from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
+from models import Avatar, ChatAttachment, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
 from services.boxim_client import BoxIMError
+from services.boxim_image_service import DownloadedBoxIMImage
 from services.takeover_service import (
     AVATAR_LOCAL_ID_PREFIX,
     TakeoverService,
@@ -83,6 +85,29 @@ class ConcurrentPollingBoxIM(FakeBoxIM):
         return []
 
 
+class ConcurrentMessagePollingBoxIM(ConcurrentPollingBoxIM):
+    async def fetch_private_messages(self, access_token, min_id="0"):
+        await super().fetch_private_messages(access_token, min_id)
+        owner_id = 100 if access_token == "prod-huihui-token" else 101
+        return [
+            {
+                "id": owner_id,
+                "localId": owner_id,
+                "sendId": owner_id + 100,
+                "recvId": owner_id,
+                "sendTime": 1_700_000_000_000,
+                "type": 0,
+                "content": "并发写入测试",
+            }
+        ]
+
+    async def mark_private_messages_read(self, access_token, friend_id, message_id):
+        await asyncio.sleep(0.05)
+        self.read_receipts.append(
+            {"friendId": str(friend_id), "messageId": str(message_id)}
+        )
+
+
 @pytest.fixture
 def service_context(tmp_path):
     engine = create_engine(
@@ -171,6 +196,247 @@ async def test_incoming_message_is_prepared_then_sent_at_three_seconds(service_c
         db.close()
 
 
+@pytest.mark.asyncio
+async def test_incoming_image_is_analyzed_and_used_in_takeover_reply(service_context):
+    session_factory, service, boxim, clock = service_context
+    await service.poll_and_process_messages()
+    boxim.messages.append(
+        {
+            "id": 111,
+            "localId": 111,
+            "sendId": 200,
+            "recvId": 100,
+            "sendTime": clock.millis(),
+            "type": 1,
+            "content": json.dumps(
+                {
+                    "originUrl": "https://cdn.example/case.png",
+                    "thumbUrl": "https://cdn.example/case-thumb.png",
+                }
+            ),
+        }
+    )
+
+    await service.poll_and_process_messages()
+    db = session_factory()
+    try:
+        scheduled = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
+        assert scheduled.status == "pending"
+        assert scheduled.prompt == "请看看这张图片。"
+    finally:
+        db.close()
+
+    clock.advance(3)
+
+    def analyze(db, avatar, content, **kwargs):
+        assert content == b"image-content"
+        attachment = ChatAttachment(
+            avatar_id=avatar.id,
+            uploader_kind=kwargs["uploader_kind"],
+            filename=kwargs["filename"],
+            mime_type="image/jpeg",
+            file_size=len(content),
+            status="ready",
+            category="medical_document",
+            summary="一张门诊病例",
+            extracted_text="主诉：咳嗽三天",
+            structured_data={"medical": {"chief_complaint": "咳嗽三天"}},
+            warning="请核对原始资料",
+            expires_at=clock.now() + timedelta(hours=24),
+        )
+        db.add(attachment)
+        db.commit()
+        db.refresh(attachment)
+        return attachment
+
+    downloaded = DownloadedBoxIMImage(
+        content=b"image-content",
+        filename="case.png",
+        mime_type="image/png",
+        source_url="https://cdn.example/case.png",
+    )
+    with (
+        patch("services.takeover_service.download_boxim_image", return_value=downloaded),
+        patch("routers.chat._analyze_image_bytes", side_effect=analyze) as analyzer,
+        patch("routers.chat._resolve_reply", return_value={"answer": "这份资料里写的是咳嗽三天。"}) as resolver,
+    ):
+        await service.poll_and_process_messages()
+
+    analyzer.assert_called_once()
+    assert resolver.call_args.args[2] == "请看看这张图片。"
+    image_contexts = resolver.call_args.kwargs["image_contexts"]
+    assert image_contexts[0]["summary"] == "一张门诊病例"
+    assert image_contexts[0]["extractedText"] == "主诉：咳嗽三天"
+    assert [item["content"] for item in boxim.sent] == ["这份资料里写的是咳嗽三天。"]
+
+    db = session_factory()
+    try:
+        event = db.query(TakeoverMessage).filter_by(boxim_message_id="111").one()
+        task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
+        assert event.attachment_id
+        assert db.get(ChatAttachment, event.attachment_id).uploader_kind == "boxim"
+        assert task.status == "sent"
+        with patch(
+            "services.takeover_service.download_boxim_image",
+            side_effect=AssertionError("cached image must not be downloaded again"),
+        ):
+            cached = service._takeover_image_attachment(db, db.get(Avatar, "avatar-1"), event)
+        assert cached.id == event.attachment_id
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_followup_text_recovers_recent_image_recorded_without_task(service_context):
+    session_factory, service, boxim, clock = service_context
+    await service.poll_and_process_messages()
+    image_message = {
+        "id": 113,
+        "localId": 113,
+        "sendId": 200,
+        "recvId": 100,
+        "sendTime": clock.millis(),
+        "type": 1,
+        "content": json.dumps(
+            {
+                "originUrl": "https://cdn.example/case.png",
+                "thumbUrl": "https://cdn.example/case-thumb.png",
+            }
+        ),
+    }
+
+    db = session_factory()
+    try:
+        avatar = db.get(Avatar, "avatar-1")
+        service._record_message(db, avatar, "100", image_message, schedule_reply=False)
+        cursor = db.query(TakeoverCursor).one()
+        cursor.last_message_id = "113"
+        db.commit()
+    finally:
+        db.close()
+
+    clock.advance(60)
+    boxim.messages.extend(
+        [
+            image_message,
+            {
+                "id": 114,
+                "localId": 114,
+                "sendId": 200,
+                "recvId": 100,
+                "sendTime": clock.millis(),
+                "type": 0,
+                "content": "请帮我看看这张图",
+            },
+        ]
+    )
+    await service.poll_messages()
+
+    db = session_factory()
+    try:
+        task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="114").one()
+        assert task.source_message_ids == ["113", "114"]
+        assert task.prompt == "请看看这张图片。\n请帮我看看这张图"
+    finally:
+        db.close()
+
+    clock.advance(1)
+    boxim.messages.append(
+        {
+            "id": 115,
+            "localId": 115,
+            "sendId": 200,
+            "recvId": 100,
+            "sendTime": clock.millis(),
+            "type": 0,
+            "content": "图里写了什么",
+        }
+    )
+    await service.poll_messages()
+
+    db = session_factory()
+    try:
+        latest = db.query(TakeoverReplyTask).filter_by(trigger_message_id="115").one()
+        assert latest.source_message_ids == ["113", "114", "115"]
+        assert latest.source_message_ids.count("113") == 1
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_explicit_followup_reuses_handled_image_within_two_days(service_context):
+    session_factory, service, boxim, clock = service_context
+    await service.poll_and_process_messages()
+    image_message = {
+        "id": 116,
+        "localId": 116,
+        "sendId": 200,
+        "recvId": 100,
+        "sendTime": clock.millis(),
+        "type": 1,
+        "content": json.dumps({"originUrl": "https://cdn.example/handled-case.png"}),
+    }
+    boxim.messages.append(image_message)
+    await service.poll_messages()
+
+    db = session_factory()
+    try:
+        image_task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="116").one()
+        image_task.status = "sent"
+        image_task.sent_at = clock.now()
+        db.commit()
+    finally:
+        db.close()
+
+    clock.advance(47 * 60 * 60)
+    boxim.messages.append(
+        {
+            "id": 117,
+            "localId": 117,
+            "sendId": 200,
+            "recvId": 100,
+            "sendTime": clock.millis(),
+            "type": 0,
+            "content": "重新看一下刚才那张病例图片",
+        }
+    )
+    await service.poll_messages()
+
+    db = session_factory()
+    try:
+        task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="117").one()
+        assert task.source_message_ids == ["116", "117"]
+        assert task.prompt == "请看看这张图片。\n重新看一下刚才那张病例图片"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_invalid_image_message_is_recorded_but_not_scheduled(service_context):
+    session_factory, service, boxim, clock = service_context
+    await service.poll_and_process_messages()
+    boxim.messages.append(
+        {
+            "id": 112,
+            "localId": 112,
+            "sendId": 200,
+            "recvId": 100,
+            "sendTime": clock.millis(),
+            "type": 1,
+            "content": json.dumps({"width": 100, "height": 100}),
+        }
+    )
+
+    await service.poll_and_process_messages()
+
+    db = session_factory()
+    try:
+        assert db.query(TakeoverMessage).filter_by(boxim_message_id="112").one()
+        assert db.query(TakeoverReplyTask).count() == 0
+    finally:
+        db.close()
+
+
 @pytest.mark.asyncio
 async def test_default_reply_delay_is_three_minutes(service_context):
     session_factory, service, boxim, clock = service_context
@@ -230,7 +496,7 @@ async def test_multiple_avatar_owners_are_polled_concurrently(service_context):
     finally:
         db.close()
 
-    boxim = ConcurrentPollingBoxIM()
+    boxim = ConcurrentMessagePollingBoxIM()
     service = TakeoverService(
         session_factory,
         boxim,
@@ -244,6 +510,8 @@ async def test_multiple_avatar_owners_are_polled_concurrently(service_context):
     db = session_factory()
     try:
         assert db.query(TakeoverCursor).filter(TakeoverCursor.initialized.is_(True)).count() == 2
+        assert db.query(TakeoverMessage).count() == 2
+        assert len(boxim.read_receipts) == 2
     finally:
         db.close()
 

digital-avatar-app/docs/H5_PRODUCTION_DEPLOYMENT.md

@@ -117,7 +117,7 @@ location /api/ {
     proxy_set_header X-Forwarded-Proto $scheme;
     proxy_buffering off;
     proxy_read_timeout 300s;
-    client_max_body_size 20m;
+    client_max_body_size 100m;
 }
 ```
 

digital-avatar-app/nginx.conf

@@ -23,6 +23,10 @@ http {
 
         root /usr/share/nginx/html;
         index index.html;
+        # Keep the application gateway aligned with the production edge gateway.
+        # Without this Nginx rejects ordinary PDF uploads with HTTP 413 before
+        # FastAPI can return its user-facing file-size validation message.
+        client_max_body_size 100m;
 
         # SPA 兜底（hash 路由下深链接也可正常加载）
         location / {

digital-avatar-app/src/api/index.ts

@@ -305,6 +305,7 @@ export interface KnowledgeDoc {
   vectorized?: boolean
   embeddingModel?: string
   chunkCount?: number
+  errorMessage?: string
   createdAt: string
 }
 
@@ -335,7 +336,8 @@ export const uploadKnowledgeDoc = (avatarId: string, file: File) => {
   const form = new FormData()
   form.append('file', file)
   return request.post<KnowledgeDoc>(`/avatar/${avatarId}/knowledge/docs`, form, {
-    headers: { 'Content-Type': 'multipart/form-data' }
+    headers: { 'Content-Type': 'multipart/form-data' },
+    timeout: 120000
   })
 }
 
@@ -343,6 +345,9 @@ export const uploadKnowledgeDoc = (avatarId: string, file: File) => {
 export const deleteKnowledgeDoc = (avatarId: string, docId: string) =>
   request.delete(`/avatar/${avatarId}/knowledge/docs/${docId}`)
 
+export const retryKnowledgeDoc = (avatarId: string, docId: string) =>
+  request.post<KnowledgeDoc>(`/avatar/${avatarId}/knowledge/docs/${docId}/retry`)
+
 // 标准问答对列表
 export const getQAPairs = (avatarId: string) =>
   request.get<QAPair[]>(`/avatar/${avatarId}/knowledge/qa`)

digital-avatar-app/src/views/KnowledgeManage.vue

@@ -27,7 +27,7 @@
             <p class="upload-hint">支持 MD / TXT / PDF / DOC / DOCX / XLSX，上传后自动向量化</p>
             <input ref="fileInput" type="file" accept=".md,.txt,.pdf,.doc,.docx,.xlsx" class="hidden-input" @change="onFileChange" />
           </div>
-          <p v-if="uploading" class="uploading-text">上传并向量化中…</p>
+          <p v-if="uploading" class="uploading-text">文件上传中…</p>
           <p v-if="uploadError" class="error-text">{{ uploadError }}</p>
         </div>
 
@@ -42,7 +42,10 @@
               <p class="card-meta">{{ doc.fileType.toUpperCase() }} · {{ formatSize(doc.fileSize) }} · {{ formatDate(doc.createdAt) }}</p>
               <p class="card-detail">{{ documentState(doc).detail }}</p>
             </div>
+            <div class="card-actions">
+              <button v-if="documentState(doc).tone === 'failed'" class="card-retry" @click="retryDoc(doc.id)">重新索引</button>
               <button class="card-delete" @click="removeDoc(doc.id)">删除</button>
+            </div>
           </article>
         </div>
         <div v-else class="card-empty">📂 暂无文档，先上传一个知识文件</div>
@@ -78,7 +81,7 @@
 </template>
 
 <script setup lang="ts">
-import { ref, onMounted, computed } from 'vue'
+import { ref, onMounted, onUnmounted, computed } from 'vue'
 import { useRoute, useRouter } from 'vue-router'
 import { useAvatarStore } from '@/store/avatar'
 import { pickScopedAvatarId, unwrapListData } from '@/utils/avatar-page-data.js'
@@ -87,6 +90,7 @@ import {
   getKnowledgeDocs,
   uploadKnowledgeDoc,
   deleteKnowledgeDoc,
+  retryKnowledgeDoc,
   getQAPairs,
   deleteQAPair,
   searchKnowledge,
@@ -107,6 +111,7 @@ const uploading = ref(false)
 const uploadError = ref('')
 const dragOver = ref(false)
 const fileInput = ref<HTMLInputElement | null>(null)
+let documentPollingTimer: ReturnType<typeof setInterval> | undefined
 
 const query = ref('')
 const searching = ref(false)
@@ -123,7 +128,26 @@ const documentState = (doc: any) => {
   if (['uploaded', 'parsing'].includes(String(doc.status || '').toLowerCase())) {
     return { tone: 'pending', label: '处理中', detail: '正在解析并建立知识索引' }
   }
-  return { tone: 'failed', label: '处理失败', detail: '未能建立知识索引，请删除后重新上传' }
+  return { tone: 'failed', label: '处理失败', detail: doc.errorMessage || '未能建立知识索引，请重新索引或重新上传' }
+}
+
+const hasPendingDocuments = () => docs.value.some((doc) =>
+  ['uploaded', 'parsing'].includes(String(doc.status || '').toLowerCase())
+)
+
+const stopDocumentPolling = () => {
+  if (documentPollingTimer) {
+    clearInterval(documentPollingTimer)
+    documentPollingTimer = undefined
+  }
+}
+
+const startDocumentPolling = () => {
+  if (documentPollingTimer || !hasPendingDocuments()) return
+  documentPollingTimer = setInterval(async () => {
+    await loadDocs()
+    if (!hasPendingDocuments()) stopDocumentPolling()
+  }, 2000)
 }
 
 const loadDocs = async () => {
@@ -131,6 +155,7 @@ const loadDocs = async () => {
   try {
     const res: any = await getKnowledgeDocs(avatarId.value)
     docs.value = unwrapListData(res)
+    startDocumentPolling()
   } catch (e) {
     console.error(e)
   }
@@ -182,6 +207,17 @@ const doUpload = async (file: File) => {
   }
 }
 
+const retryDoc = async (id: string) => {
+  if (!avatarId.value) return
+  uploadError.value = ''
+  try {
+    await retryKnowledgeDoc(avatarId.value, id)
+    await loadDocs()
+  } catch (e: any) {
+    uploadError.value = e?.message || '重新索引失败'
+  }
+}
+
 const removeDoc = async (id: string) => {
   if (!avatarId.value) return
   await deleteKnowledgeDoc(avatarId.value, id)
@@ -257,6 +293,8 @@ onMounted(async () => {
   if (avatarId.value) store.currentAvatarId = avatarId.value
   await Promise.all([loadDocs(), loadQA()])
 })
+
+onUnmounted(stopDocumentPolling)
 </script>
 
 <style scoped>
@@ -300,7 +338,10 @@ onMounted(async () => {
 .status-pill.missing { color: #B91C1C; background: #FEF2F2; }
 .status-pill.failed { color: #B91C1C; background: #FEF2F2; }
 .card-meta, .card-detail { margin: 5px 0 0; color: #9398AE; font-size: 11px; line-height: 1.4; }.card-detail { color: #8B6B58; }
-.card-delete { flex: 0 0 auto; align-self: center; border: 0; color: #EF4444; background: #FEF2F2; border-radius: 8px; padding: 7px 9px; font-size: 12px; cursor: pointer; }
+.card-actions { flex: 0 0 auto; display: flex; flex-direction: column; align-items: stretch; gap: 6px; }
+.card-delete, .card-retry { align-self: center; border: 0; border-radius: 8px; padding: 7px 9px; font-size: 12px; cursor: pointer; white-space: nowrap; }
+.card-delete { color: #EF4444; background: #FEF2F2; }
+.card-retry { color: #C15F18; background: #FFF3E6; }
 .card-empty { padding: 42px 16px; border: 1px dashed #F1D9C3; border-radius: 16px; color: #9398AE; background: #fff; font-size: 14px; text-align: center; }
 .qa-card { align-items: stretch; text-align: left; }.qa-card.qa-disabled { opacity: .58; }
 .qa-card .card-content,