fix(backend): 修复全面审阅发现的核心问题

修复索引首次失败回滚、Vault 扫描边界、Markdown 代码围栏和 UTF-16 Citation 偏移。

收紧 Plugin 权限与 JSON Schema 校验，补齐 Note Move、Task、Attachment 和 Transcript Tool。

接入 OpenAI SSE 与 Ollama JSONL 真流式输出，修正 Provider PATCH 语义并限制运行时内存保留。

新增对应回归测试，后端测试增至 62 项。

backend/app/services/attachment_service.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from app.config import get_settings
+from app.errors import ApiError
+
+_ATTACHMENT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$")
+MAX_ATTACHMENT_BYTES = 1024 * 1024
+
+
+def attachment_path(attachment_id: str) -> Path:
+    if not _ATTACHMENT_ID.fullmatch(attachment_id):
+        raise ApiError(
+            400, "INVALID_ATTACHMENT_ID", "attachment_id is invalid",
+            {"attachment_id": attachment_id},
+        )
+    root = get_settings().attachments_path.resolve()
+    candidate = (root / attachment_id).resolve()
+    if not candidate.is_relative_to(root):
+        raise ApiError(400, "INVALID_ATTACHMENT_ID", "attachment path escapes storage")
+    return candidate
+
+
+def read_attachment(attachment_id: str, *, max_chars: int = 100_000) -> dict[str, object]:
+    path = attachment_path(attachment_id)
+    if not path.is_file():
+        raise ApiError(
+            404, "ATTACHMENT_NOT_FOUND", "attachment not found",
+            {"attachment_id": attachment_id},
+        )
+    size = path.stat().st_size
+    if size > MAX_ATTACHMENT_BYTES:
+        raise ApiError(
+            413, "ATTACHMENT_TOO_LARGE", "attachment exceeds the Tool read limit",
+            {"attachment_id": attachment_id, "size": size},
+        )
+    text = path.read_text(encoding="utf-8")
+    truncated = len(text) > max_chars
+    return {
+        "attachment_id": attachment_id,
+        "content": text[:max_chars],
+        "size": size,
+        "truncated": truncated,
+    }

backend/app/services/coordination.py

@@ -0,0 +1,15 @@
+import asyncio
+from functools import wraps
+
+_vault_mutation_lock = asyncio.Lock()
+
+
+def serialized_vault_mutation(operation):
+    """串行化 Vault 文件与可重建索引的写入，避免 rebuild 与 Note 写操作交错。"""
+
+    @wraps(operation)
+    async def wrapped(*args, **kwargs):
+        async with _vault_mutation_lock:
+            return await operation(*args, **kwargs)
+
+    return wrapped

backend/app/services/index_service.py

@@ -17,11 +17,24 @@ from app.contracts import IndexJob, IndexRebuildRequest, IndexStatus
 from app.errors import ApiError
 from app.knowledge.parser import parse_note
 from app.services.note_service import index_note
+from app.services import task_service
+from app.services.coordination import serialized_vault_mutation
 from app.retrieval.vectorstore import SqliteVecStore
 
 vector_store = SqliteVecStore()
 
 _jobs: dict[str, IndexJob] = {}
+_active_job_id: str | None = None
+_last_completed_at: datetime | None = None
+_last_error: str | None = None
+MAX_JOBS = 100
+
+
+def _remember_job(job: IndexJob) -> None:
+    _jobs[job.job_id] = job
+    while len(_jobs) > MAX_JOBS:
+        oldest = next(iter(_jobs))
+        _jobs.pop(oldest, None)
 
 
 def _scan_vault() -> list[tuple[str, str, str, datetime, datetime]]:
@@ -29,23 +42,28 @@ def _scan_vault() -> list[tuple[str, str, str, datetime, datetime]]:
 
     先读入内存：若文件读取失败，rebuild 尚未清空旧索引，不会造成数据损失。
     """
-    vault = get_settings().vault_path
+    vault = get_settings().vault_path.resolve()
     result: list[tuple[str, str, str, datetime, datetime]] = []
     if not vault.exists():
         return result
     for path in sorted(vault.rglob("*.md")):
+        resolved = path.resolve()
+        if not resolved.is_relative_to(vault):
+            continue
         rel = path.relative_to(vault).as_posix()
         folder = path.relative_to(vault).parent.as_posix()
         if folder == ".":
             folder = ""
-        stat = path.stat()
+        stat = resolved.stat()
         created = datetime.fromtimestamp(stat.st_ctime, tz=timezone.utc)
         updated = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
-        result.append((rel, folder, path.read_text(encoding="utf-8"), created, updated))
+        result.append((rel, folder, resolved.read_text(encoding="utf-8"), created, updated))
     return result
 
 
+@serialized_vault_mutation
 async def rebuild(request: IndexRebuildRequest) -> IndexJob:
+    global _active_job_id, _last_completed_at, _last_error
     job_id = "job_" + uuid4().hex[:12]
     # 增量重建（scope != all 或指定 note_ids）尚未实现，明确拒绝而非静默全量重建
     if request.scope != "all" or request.note_ids:
@@ -59,10 +77,22 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob:
     # 先扫描到内存（失败不会清旧索引），再快照旧库用于失败回滚
     docs = _scan_vault()
     settings = get_settings()
-    backup_path = settings.db_path.with_suffix(".db.bak") if settings.db_path.exists() else None
+    database_existed = settings.db_path.exists()
+    task_note_links = task_service.note_links() if database_existed else {}
+    backup_path = (
+        settings.db_path.with_name(f"{settings.db_path.name}.{job_id}.bak")
+        if database_existed
+        else None
+    )
     if backup_path is not None:
         shutil.copy2(settings.db_path, backup_path)
 
+    _active_job_id = job_id
+    _last_error = None
+    _remember_job(IndexJob(
+        job_id=job_id, status="running", scope=request.scope,
+        created_at=datetime.now(timezone.utc),
+    ))
     try:
         repository.clear_all()
         await vector_store.clear()
@@ -72,27 +102,39 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob:
                 created_at=created, updated_at=updated,
             )
             await index_note(parsed)
-    except BaseException:
+        task_service.restore_note_links(task_note_links)
+    except BaseException as exc:
         # 重建失败：恢复旧索引，避免留下半成品；记录 failed 任务后向上抛
         if backup_path is not None and backup_path.exists():
             shutil.copy2(backup_path, settings.db_path)
-        _jobs[job_id] = IndexJob(
+        elif not database_existed:
+            settings.db_path.unlink(missing_ok=True)
+        _remember_job(IndexJob(
             job_id=job_id, status="failed", scope=request.scope,
             created_at=datetime.now(timezone.utc),
-        )
+        ))
+        _last_error = str(exc)
         raise
     finally:
+        _active_job_id = None
         if backup_path is not None:
             backup_path.unlink(missing_ok=True)
 
     job = IndexJob(job_id=job_id, status="completed", scope=request.scope, created_at=datetime.now(timezone.utc))
-    _jobs[job_id] = job
+    _remember_job(job)
+    _last_completed_at = job.created_at
     return job
 
 
 def get_status() -> IndexStatus:
-    # 同步重建、无排队任务，因此状态恒为 idle；实际索引规模可由 GET /api/notes 与搜索反映
-    return IndexStatus(status="idle", pending_jobs=0)
+    if _active_job_id is not None:
+        return IndexStatus(status="running", pending_jobs=0, active_job_id=_active_job_id)
+    return IndexStatus(
+        status="failed" if _last_error else "idle",
+        pending_jobs=0,
+        last_completed_at=_last_completed_at,
+        error_message=_last_error,
+    )
 
 
 def get_job(job_id: str) -> IndexJob | None:

backend/app/services/note_service.py

@@ -19,6 +19,7 @@ from app.errors import ApiError
 from app.knowledge.parser import ParsedNote, parse_note
 from app.retrieval.embedding import HashEmbeddingProvider
 from app.retrieval.vectorstore import SqliteVecStore, VectorRecord
+from app.services.coordination import serialized_vault_mutation
 
 # 轻量实现实例（无状态，可直接复用）；接入真实模型后替换为对应 Provider
 embedding = HashEmbeddingProvider()
@@ -148,6 +149,7 @@ async def index_note(parsed: ParsedNote) -> None:
         conn.close()
 
 
+@serialized_vault_mutation
 async def create_note(*, title: str, markdown: str, folder: str | None, tags: list[str]) -> Note:
     rel_path, clean_folder = _rel_path(folder, title)
     now = datetime.now(timezone.utc)
@@ -183,6 +185,7 @@ async def get_note(note_id: str) -> Note | None:
                        record.created_at, record.updated_at, record.blocks, markdown)
 
 
+@serialized_vault_mutation
 async def update_note(
     note_id: str, *, title: str | None = None, markdown: str | None = None, tags: list[str] | None = None
 ) -> Note:
@@ -213,6 +216,58 @@ async def update_note(
                        parsed.created_at, parsed.updated_at, parsed.blocks, new_md)
 
 
+@serialized_vault_mutation
+async def move_note(note_id: str, *, folder: str) -> Note:
+    record = repository.get_note_record(note_id)
+    if record is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
+
+    clean_folder = _normalize_folder(folder)
+    filename = Path(record.file_path).name
+    new_rel_path = f"{clean_folder}/{filename}" if clean_folder else filename
+    if new_rel_path == record.file_path:
+        note = await get_note(note_id)
+        assert note is not None
+        return note
+
+    source = _abs_path(record.file_path)
+    target = _abs_path(new_rel_path)
+    if not source.is_file():
+        raise ApiError(
+            409, "NOTE_FILE_MISSING", "note file is missing from the Vault",
+            {"note_id": note_id, "file_path": record.file_path},
+        )
+    if target.exists():
+        raise ApiError(
+            409, "RESOURCE_CONFLICT", "a note already exists at the target path",
+            {"note_id": note_id, "file_path": new_rel_path},
+        )
+
+    markdown = source.read_text(encoding="utf-8")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    source.replace(target)
+    try:
+        parsed = parse_note(
+            markdown=markdown,
+            file_path=new_rel_path,
+            folder=clean_folder,
+            tags=record.tags,
+            created_at=record.created_at,
+            updated_at=datetime.now(timezone.utc),
+            note_id=record.note_id,
+        )
+        parsed.title = record.title
+        await index_note(parsed)
+    except BaseException:
+        target.replace(source)
+        raise
+    return _build_note(
+        parsed.note_id, parsed.title, parsed.file_path, parsed.tags,
+        parsed.created_at, parsed.updated_at, parsed.blocks, markdown,
+    )
+
+
+@serialized_vault_mutation
 async def delete_note(note_id: str) -> bool:
     record = repository.get_note_record(note_id)
     if record is None:

backend/app/services/task_service.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from app import repository
+from app.contracts import Task, TaskStatus
+from app.database.db import connect, transaction
+from app.errors import ApiError
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _task_from_row(row) -> Task:
+    return Task(
+        task_id=row["task_id"],
+        title=row["title"],
+        description=row["description"],
+        status=TaskStatus(row["status"]),
+        note_id=row["note_id"],
+        due_at=datetime.fromisoformat(row["due_at"]) if row["due_at"] else None,
+        created_at=datetime.fromisoformat(row["created_at"]),
+        updated_at=datetime.fromisoformat(row["updated_at"]),
+    )
+
+
+def create_task(
+    *, title: str, description: str = "", note_id: str | None = None,
+    due_at: datetime | None = None,
+) -> Task:
+    if note_id and repository.get_note_record(note_id) is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
+    task_id = f"task_{uuid4().hex}"
+    now = _now()
+    conn = connect()
+    try:
+        with transaction(conn):
+            conn.execute(
+                """
+                INSERT INTO tasks
+                    (task_id, title, description, status, note_id, due_at, created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    task_id, title, description, TaskStatus.todo.value, note_id,
+                    due_at.isoformat() if due_at else None, now.isoformat(), now.isoformat(),
+                ),
+            )
+        row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
+        return _task_from_row(row)
+    finally:
+        conn.close()
+
+
+def get_task(task_id: str) -> Task | None:
+    conn = connect()
+    try:
+        row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
+        return _task_from_row(row) if row else None
+    finally:
+        conn.close()
+
+
+def list_tasks(*, limit: int, offset: int) -> tuple[list[Task], int]:
+    conn = connect()
+    try:
+        total = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
+        rows = conn.execute(
+            "SELECT * FROM tasks ORDER BY updated_at DESC LIMIT ? OFFSET ?",
+            (limit, offset),
+        ).fetchall()
+        return [_task_from_row(row) for row in rows], total
+    finally:
+        conn.close()
+
+
+def update_task(task_id: str, values: dict[str, object]) -> Task:
+    current = get_task(task_id)
+    if current is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "task not found", {"task_id": task_id})
+    if "note_id" in values and values["note_id"]:
+        note_id = str(values["note_id"])
+        if repository.get_note_record(note_id) is None:
+            raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
+    if values.get("title") is None:
+        values.pop("title", None)
+    if values.get("description") is None:
+        values.pop("description", None)
+    if values.get("status") is None:
+        values.pop("status", None)
+
+    columns: list[str] = []
+    params: list[object] = []
+    for name, value in values.items():
+        columns.append(f"{name} = ?")
+        if isinstance(value, datetime):
+            value = value.isoformat()
+        elif isinstance(value, TaskStatus):
+            value = value.value
+        params.append(value)
+    columns.append("updated_at = ?")
+    params.append(_now().isoformat())
+    params.append(task_id)
+    conn = connect()
+    try:
+        with transaction(conn):
+            conn.execute(
+                f"UPDATE tasks SET {', '.join(columns)} WHERE task_id = ?",
+                params,
+            )
+        row = conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)).fetchone()
+        return _task_from_row(row)
+    finally:
+        conn.close()
+
+
+def delete_task(task_id: str) -> bool:
+    conn = connect()
+    try:
+        with transaction(conn):
+            cursor = conn.execute("DELETE FROM tasks WHERE task_id = ?", (task_id,))
+        return cursor.rowcount > 0
+    finally:
+        conn.close()
+
+
+def note_links() -> dict[str, str]:
+    """重建可再生 Note 索引前，暂存不可再生 Task 到 Note 的业务关联。"""
+    conn = connect()
+    try:
+        return {
+            row["task_id"]: row["note_id"]
+            for row in conn.execute(
+                "SELECT task_id, note_id FROM tasks WHERE note_id IS NOT NULL"
+            )
+        }
+    finally:
+        conn.close()
+
+
+def restore_note_links(links: dict[str, str]) -> None:
+    if not links:
+        return
+    conn = connect()
+    try:
+        with transaction(conn):
+            for task_id, note_id in links.items():
+                exists = conn.execute(
+                    "SELECT 1 FROM notes WHERE note_id = ?", (note_id,)
+                ).fetchone()
+                if exists:
+                    conn.execute(
+                        "UPDATE tasks SET note_id = ? WHERE task_id = ?",
+                        (note_id, task_id),
+                    )
+    finally:
+        conn.close()

backend/app/services/transcription_service.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from collections import OrderedDict
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+from app.contracts import TranscriptionJob
+from app.services.attachment_service import attachment_path
+
+_jobs: OrderedDict[str, TranscriptionJob] = OrderedDict()
+MAX_JOBS = 100
+
+
+def create_transcription(attachment_id: str, language: str | None = None) -> TranscriptionJob:
+    del language  # 预生成 transcript 暂不需要语言识别。
+    source = attachment_path(attachment_id)
+    transcript = source if source.suffix.lower() in {".txt", ".md"} else Path(f"{source}.txt")
+    job = TranscriptionJob(
+        job_id=f"transcription_{uuid4().hex}",
+        attachment_id=attachment_id,
+        status="completed" if transcript.is_file() else "failed",
+        text=transcript.read_text(encoding="utf-8") if transcript.is_file() else None,
+        error_code=None if transcript.is_file() else "TRANSCRIPTION_BACKEND_UNAVAILABLE",
+        error_message=(
+            None
+            if transcript.is_file()
+            else "No host-generated transcript is available; local speech models are phase two."
+        ),
+        created_at=datetime.now(timezone.utc),
+    )
+    _jobs[job.job_id] = job
+    while len(_jobs) > MAX_JOBS:
+        _jobs.popitem(last=False)
+    return job.model_copy(deep=True)
+
+
+def get_transcription(job_id: str) -> TranscriptionJob | None:
+    job = _jobs.get(job_id)
+    return job.model_copy(deep=True) if job else None