diff --git a/backend/app/routes.py b/backend/app/routes.py
index 7f5db3f..3eca5e9 100644
--- a/backend/app/routes.py
+++ b/backend/app/routes.py
@@ -51,6 +51,8 @@ from app.container import container
 from app.errors import ApiError, not_implemented
 from app.providers.registry import ProviderNotFoundError
 from app.providers.factory import UnsupportedProviderError
+from app.retrieval.engine import engine
+from app.services import index_service, note_service
 
 router = APIRouter(prefix="/api")
 not_implemented_response = {501: {"model": ErrorResponse, "description": "业务服务尚未实现"}}
@@ -108,38 +110,37 @@ async def list_notes(
     folder: str | None = None,
     tag: str | None = None,
 ) -> NoteListResponse:
-    return NoteListResponse(page=PageMeta(limit=limit, offset=offset))
+    items, total = note_service.list_notes(limit=limit, offset=offset, folder=folder, tag=tag)
+    return NoteListResponse(items=items, page=PageMeta(total=total, limit=limit, offset=offset))
 
 
-@router.post(
-    "/notes", response_model=Note, responses=not_implemented_response, tags=["Notes"]
-)
-async def create_note(_: NoteCreateRequest) -> Note:
-    not_implemented("notes.create")
+@router.post("/notes", response_model=Note, tags=["Notes"])
+async def create_note(request: NoteCreateRequest) -> Note:
+    return await note_service.create_note(
+        title=request.title, markdown=request.markdown, folder=request.folder, tags=request.tags
+    )
 
 
-@router.get(
-    "/notes/{note_id}", response_model=Note, responses=not_implemented_response, tags=["Notes"]
-)
+@router.get("/notes/{note_id}", response_model=Note, tags=["Notes"])
 async def get_note(note_id: str) -> Note:
-    not_implemented(f"notes.read:{note_id}")
+    note = await note_service.get_note(note_id)
+    if note is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
+    return note
 
 
-@router.patch(
-    "/notes/{note_id}", response_model=Note, responses=not_implemented_response, tags=["Notes"]
-)
-async def update_note(note_id: str, _: NoteUpdateRequest) -> Note:
-    not_implemented(f"notes.update:{note_id}")
+@router.patch("/notes/{note_id}", response_model=Note, tags=["Notes"])
+async def update_note(note_id: str, request: NoteUpdateRequest) -> Note:
+    return await note_service.update_note(
+        note_id, title=request.title, markdown=request.markdown, tags=request.tags
+    )
 
 
-@router.delete(
-    "/notes/{note_id}",
-    response_model=OperationResponse,
-    responses=not_implemented_response,
-    tags=["Notes"],
-)
+@router.delete("/notes/{note_id}", response_model=OperationResponse, tags=["Notes"])
 async def delete_note(note_id: str) -> OperationResponse:
-    not_implemented(f"notes.delete:{note_id}")
+    if not await note_service.delete_note(note_id):
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id})
+    return OperationResponse(status="completed", resource_id=note_id, message="deleted")
 
 
 @router.post(
@@ -152,11 +153,7 @@ async def move_note(note_id: str, _: NoteMoveRequest) -> Note:
 # Retrieval and chat
 @router.post("/search", response_model=SearchResponse, tags=["Search"])
 async def search_notes(request: SearchRequest) -> SearchResponse:
-    return SearchResponse(
-        query=request.query,
-        mode=request.mode,
-        page=PageMeta(limit=request.limit, offset=request.offset),
-    )
+    return await engine.search(request)
 
 
 @router.post(
@@ -576,25 +573,22 @@ async def get_transcription(job_id: str) -> TranscriptionJob:
 
 @router.get("/index/status", response_model=IndexStatus, tags=["Index"])
 async def get_index_status() -> IndexStatus:
-    return IndexStatus()
+    return index_service.get_status()
 
 
 @router.post(
     "/index/rebuild",
     response_model=IndexJob,
     status_code=202,
-    responses=not_implemented_response,
     tags=["Index"],
 )
-async def rebuild_index(_: IndexRebuildRequest) -> IndexJob:
-    not_implemented("index.rebuild")
+async def rebuild_index(request: IndexRebuildRequest) -> IndexJob:
+    return await index_service.rebuild(request)
 
 
-@router.get(
-    "/index/jobs/{job_id}",
-    response_model=IndexJob,
-    responses=not_implemented_response,
-    tags=["Index"],
-)
+@router.get("/index/jobs/{job_id}", response_model=IndexJob, tags=["Index"])
 async def get_index_job(job_id: str) -> IndexJob:
-    not_implemented(f"index.jobs.read:{job_id}")
+    job = index_service.get_job(job_id)
+    if job is None:
+        raise ApiError(404, "RESOURCE_NOT_FOUND", "index job not found", {"job_id": job_id})
+    return job
diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py
new file mode 100644
index 0000000..e94fc8c
--- /dev/null
+++ b/backend/app/services/__init__.py
@@ -0,0 +1 @@
+"""应用服务层：编排 Knowledge/Retrieval Core 与 Repository，供路由调用。"""
diff --git a/backend/app/services/index_service.py b/backend/app/services/index_service.py
new file mode 100644
index 0000000..d85eb6a
--- /dev/null
+++ b/backend/app/services/index_service.py
@@ -0,0 +1,69 @@
+"""索引服务：扫描 Vault、全量重建索引、查询索引状态。
+
+MVP 阶段重建是同步的（数据量小），完成后直接返回 completed 的 IndexJob。
+索引任务暂存内存（_jobs），不持久化到 SQLite；后续接入异步任务队列时再落到 index_jobs 表。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+from app import repository
+from app.config import get_settings
+from app.contracts import IndexJob, IndexRebuildRequest, IndexStatus
+from app.knowledge.parser import parse_note
+from app.services.note_service import index_note
+from app.retrieval.vectorstore import SqliteVecStore
+
+vector_store = SqliteVecStore()
+
+_jobs: dict[str, IndexJob] = {}
+
+
+def _scan_vault() -> list[tuple[str, str, str]]:
+    """扫描 Vault 下所有 Markdown，返回 (rel_path, folder, markdown)。"""
+    vault = get_settings().vault_path
+    result: list[tuple[str, str, str]] = []
+    if not vault.exists():
+        return result
+    for path in sorted(vault.rglob("*.md")):
+        rel = path.relative_to(vault).as_posix()
+        folder = path.relative_to(vault).parent.as_posix()
+        if folder == ".":
+            folder = ""
+        result.append((rel, folder, path.read_text(encoding="utf-8")))
+    return result
+
+
+async def rebuild(request: IndexRebuildRequest) -> IndexJob:
+    job_id = "job_" + uuid4().hex[:12]
+    # MVP：scope（all/notes/vectors）与 note_ids 增量暂不区分，统一全量重建
+    repository.clear_all()
+    await vector_store.clear()
+
+    vault = get_settings().vault_path
+    for rel, folder, markdown in _scan_vault():
+        path = vault / rel
+        stat = path.stat()
+        created = datetime.fromtimestamp(stat.st_ctime, tz=timezone.utc)
+        updated = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
+        parsed = parse_note(
+            markdown=markdown, file_path=rel, folder=folder, tags=None,
+            created_at=created, updated_at=updated,
+        )
+        await index_note(parsed)
+
+    job = IndexJob(job_id=job_id, status="completed", scope=request.scope, created_at=datetime.now(timezone.utc))
+    _jobs[job_id] = job
+    return job
+
+
+def get_status() -> IndexStatus:
+    # 同步重建、无排队任务，因此状态恒为 idle；实际索引规模可由 GET /api/notes 与搜索反映
+    return IndexStatus(status="idle", pending_jobs=0)
+
+
+def get_job(job_id: str) -> IndexJob | None:
+    return _jobs.get(job_id)
diff --git a/backend/app/services/note_service.py b/backend/app/services/note_service.py
new file mode 100644
index 0000000..d3e6753
--- /dev/null
+++ b/backend/app/services/note_service.py
@@ -0,0 +1,159 @@
+"""Note 服务：Markdown 文件读写 + 解析 + 索引编排。
+
+Markdown 文件是笔记正文的持久化载体（Vault），SQLite/FTS5/向量是可重建索引。
+本服务负责在两者之间保持一致：写文件后解析并写入元数据、FTS5 与向量。
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+
+from app import repository
+from app.config import get_settings
+from app.contracts import Note, NoteBlock, NoteSummary
+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
+
+# 轻量实现实例（无状态，可直接复用）；接入真实模型后替换为对应 Provider
+embedding = HashEmbeddingProvider()
+vector_store = SqliteVecStore()
+
+
+def _vault() -> Path:
+    return get_settings().vault_path
+
+
+def _safe_name(title: str) -> str:
+    name = re.sub(r'[\\/:*?"<>|]', "_", title).strip()
+    return name or "untitled"
+
+
+def _rel_path(folder: str | None, title: str) -> str:
+    folder_part = folder.strip().strip("/") if folder else ""
+    name = _safe_name(title)
+    if not name.endswith(".md"):
+        name += ".md"
+    return f"{folder_part}/{name}" if folder_part else name
+
+
+def _abs_path(rel_path: str) -> Path:
+    return _vault() / rel_path
+
+
+def _read_markdown(rel_path: str) -> str:
+    path = _abs_path(rel_path)
+    return path.read_text(encoding="utf-8") if path.exists() else ""
+
+
+def _write_markdown(rel_path: str, markdown: str) -> None:
+    path = _abs_path(rel_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(markdown, encoding="utf-8")
+
+
+def _delete_markdown(rel_path: str) -> None:
+    path = _abs_path(rel_path)
+    if path.exists():
+        path.unlink()
+
+
+async def index_note(parsed: ParsedNote) -> None:
+    """把解析结果写入元数据 + FTS5 + 向量（三层可重建索引）。"""
+    vectors = await embedding.embed_documents([block.content for block in parsed.blocks])
+    repository.replace_note_metadata(
+        note_id=parsed.note_id,
+        title=parsed.title,
+        file_path=parsed.file_path,
+        folder=parsed.folder,
+        tags=parsed.tags,
+        created_at=parsed.created_at,
+        updated_at=parsed.updated_at,
+        blocks=parsed.blocks,
+    )
+    records = [
+        VectorRecord(id=block.block_id, vector=vector)
+        for block, vector in zip(parsed.blocks, vectors)
+    ]
+    await vector_store.upsert(records)
+    repository.set_index_meta({"embedding_model": embedding.model_id, "embedding_dim": str(embedding.dim)})
+
+
+async def create_note(*, title: str, markdown: str, folder: str | None, tags: list[str]) -> Note:
+    rel_path = _rel_path(folder, title)
+    _write_markdown(rel_path, markdown)
+    now = datetime.now(timezone.utc)
+    parsed = parse_note(
+        markdown=markdown, file_path=rel_path, folder=folder or "", tags=tags,
+        created_at=now, updated_at=now,
+    )
+    parsed.title = title  # 显式传入的 title 优先于正文推导（与 update_note 保持一致）
+    await index_note(parsed)
+    return _build_note(parsed.note_id, parsed.title, parsed.file_path, parsed.tags,
+                       parsed.created_at, parsed.updated_at, parsed.blocks, markdown)
+
+
+async def get_note(note_id: str) -> Note | None:
+    record = repository.get_note_record(note_id)
+    if record is None:
+        return None
+    markdown = _read_markdown(record.file_path)
+    return _build_note(record.note_id, record.title, record.file_path, record.tags,
+                       record.created_at, record.updated_at, record.blocks, markdown)
+
+
+async def update_note(
+    note_id: str, *, title: str | None = None, markdown: str | None = None, tags: list[str] | None = None
+) -> 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})
+
+    new_md = _read_markdown(record.file_path) if markdown is None else markdown
+    _write_markdown(record.file_path, new_md)
+
+    now = datetime.now(timezone.utc)
+    parsed = parse_note(
+        markdown=new_md, file_path=record.file_path, folder=record.folder, tags=tags,
+        created_at=record.created_at, updated_at=now,
+    )
+    if title is not None:
+        parsed.title = title  # 显式传入的 title 覆盖正文推导结果
+
+    await index_note(parsed)
+    return _build_note(parsed.note_id, parsed.title, parsed.file_path, parsed.tags,
+                       parsed.created_at, parsed.updated_at, parsed.blocks, new_md)
+
+
+async def delete_note(note_id: str) -> bool:
+    record = repository.get_note_record(note_id)
+    if record is None:
+        return False
+    block_ids = repository.delete_note(note_id)
+    await vector_store.delete(block_ids)
+    _delete_markdown(record.file_path)
+    return True
+
+
+def list_notes(*, limit: int, offset: int, folder: str | None, tag: str | None) -> tuple[list[NoteSummary], int]:
+    items, total = repository.list_note_summaries(limit=limit, offset=offset, folder=folder, tag=tag)
+    return [NoteSummary(**item) for item in items], total
+
+
+def _build_note(
+    note_id: str, title: str, file_path: str, tags: list[str],
+    created_at: datetime, updated_at: datetime, blocks: list[NoteBlock], markdown: str,
+) -> Note:
+    return Note(
+        note_id=note_id,
+        title=title,
+        file_path=file_path,
+        tags=tags,
+        created_at=created_at,
+        updated_at=updated_at,
+        markdown=markdown,
+        blocks=blocks,
+    )