diff --git a/app/services/cloud/document_text.py b/app/services/cloud/document_text.py
new file mode 100644
index 0000000..92af610
--- /dev/null
+++ b/app/services/cloud/document_text.py
@@ -0,0 +1,63 @@
+"""Cloud document full-text reader.
+
+Reads the persisted parsed text of a cloud-drive document from MongoDB
+``cloud_drive_documents`` (written by ``doc_parser.vectorize._extract_text``
+during the processing pipeline). Used by the note-conversion flow to feed a
+document's full text into the note agent — no MinIO download or re-parsing
+needed.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Upper bound on characters fed to the note agent. A single LLM prompt can
+# absorb ~50k CJK chars comfortably; larger documents are truncated (the
+# agent still produces a faithful note over the leading content).
+MAX_DOC_TEXT_CHARS = 50_000
+
+COLLECTION = "cloud_drive_documents"
+
+
+class CloudDocumentNotFoundError(Exception):
+    """No parsed document exists for this upload_uuid (wrong uid or not parsed yet)."""
+
+
+async def read_cloud_document_text(upload_uuid: str, uid: int) -> dict[str, Any]:
+    """Return ``{"file_name", "content", "parser", "truncated"}`` for a document.
+
+    Ownership is enforced by filtering on ``uid`` (stored alongside the
+    document when it was parsed). Raises :class:`CloudDocumentNotFoundError`
+    when the document is missing, not owned, or its text is empty.
+    """
+    from app.infra.mongo import get_database
+
+    db = get_database()
+    if db is None:
+        raise RuntimeError("MongoDB 未启用")
+
+    doc = await db[COLLECTION].find_one(
+        {"upload_uuid": upload_uuid, "uid": uid},
+        {"title": 1, "content": 1, "content_source": 1, "_id": 0},
+    )
+    raw_text = str((doc or {}).get("content") or "").strip()
+    if doc is None or not raw_text:
+        raise CloudDocumentNotFoundError(upload_uuid)
+
+    truncated = len(raw_text) > MAX_DOC_TEXT_CHARS
+    if truncated:
+        logger.warning(
+            "[CLOUD_DOC_TEXT] truncating upload=%s chars=%d cap=%d",
+            upload_uuid[:8],
+            len(raw_text),
+            MAX_DOC_TEXT_CHARS,
+        )
+    return {
+        "file_name": str(doc.get("title") or ""),
+        "content": raw_text[:MAX_DOC_TEXT_CHARS],
+        "parser": str(doc.get("content_source") or ""),
+        "truncated": truncated,
+    }
diff --git a/app/services/doc_parser/legacy_doc.py b/app/services/doc_parser/legacy_doc.py
new file mode 100644
index 0000000..5e3ef93
--- /dev/null
+++ b/app/services/doc_parser/legacy_doc.py
@@ -0,0 +1,191 @@
+"""Best-effort text extraction for legacy Word 97-2003 ``.doc`` files.
+
+Browsers cannot render binary ``.doc`` natively and python-docx only reads
+OOXML ``.docx``. This module provides a *preview-only* text extractor so
+legacy documents can still be viewed online.
+
+Quality caveats (why this is preview-only, never fed into RAG):
+- tables / images / headers are flattened or dropped
+- heuristic piece decoding may produce artifacts on exotic encodings
+
+Algorithm:
+1. Open the OLE2 compound document (``olefile``) and read ``WordDocument``.
+2. Parse the FIB header: ``fcMin``/``fcMac`` (simple non-complex files) and
+   ``fcClx``/``lcbClx`` (piece table location), plus ``fWhichTblStm`` to pick
+   the ``0Table``/``1Table`` stream.
+3. If a piece table (CLX) exists, walk its PLCPCD entries: each PCD carries
+   a file offset whose bit 30 marks ANSI(=8-bit, cp1252/gbk) vs UTF-16 text.
+4. Fallback for non-complex files: decode ``WordDocument[fcMin:fcMac]``
+   directly (UTF-16LE when FIB says fExtChar, else cp936/cp1252 heuristics).
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+_FCLC_COMPLEX = 0x0004  # FIB flag: piece table (CLX) is authoritative
+_FEXTCHAR_100 = 0x1000  # FIB flag (old offset): text is Unicode
+_FWHICHTBLSTM = 0x0200  # FIB flag: table stream name is "1Table" not "0Table"
+_PCD_FC_COMPRESSED = 0x40000000  # PCD fc bit 30: piece is 8-bit ANSI
+
+_CTRL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+_MULTI_BLANK_RE = re.compile(r"\n{3,}")
+_SPECIAL_RE = re.compile(r"[\r\x07\x0b\x0c\x1e\x1f]")
+
+
+def extract_doc_text(content: bytes) -> str | None:
+    """Extract best-effort text from a .doc byte blob.
+
+    Returns ``None`` when the blob is not a recognizable Word document or
+    parsing fails — callers fall back to download-only UI.
+    """
+    try:
+        import olefile
+
+        ole = olefile.OleFileIO(io.BytesIO(content))
+    except Exception:
+        return None
+
+    try:
+        if not ole.exists("WordDocument"):
+            return None
+        word_stream = ole.openstream("WordDocument").read()
+        if len(word_stream) < 0x200 or word_stream[:2] != b"\xec\xa5":
+            # wIdent magic 0xA5EC (little-endian).
+            return None
+
+        flags = int.from_bytes(word_stream[0x0A:0x0C], "little")
+        text = _extract_via_piece_table(ole, word_stream, flags)
+        if text is None:
+            text = _extract_simple(word_stream, flags)
+        if not text:
+            return None
+        return _clean(text)[:500_000]
+    except Exception:
+        logger.warning("[DOC_PREVIEW] extraction failed", exc_info=True)
+        return None
+    finally:
+        ole.close()
+
+
+def _extract_via_piece_table(
+    ole, word_stream: bytes, flags: int
+) -> str | None:
+    """Decode text through the CLX piece table (complex + most modern files)."""
+    fc_clx = int.from_bytes(word_stream[0x01A2:0x01A6], "little")
+    lcb_clx = int.from_bytes(word_stream[0x01A6:0x01AA], "little")
+    if lcb_clx == 0:
+        return None
+
+    table_name = "1Table" if flags & _FWHICHTBLSTM else "0Table"
+    if not ole.exists(table_name):
+        table_name = "1Table" if table_name == "0Table" else "0Table"
+        if not ole.exists(table_name):
+            return None
+    clx = ole.openstream(table_name).read()[fc_clx : fc_clx + lcb_clx]
+
+    pieces = _parse_clx(clx)
+    if not pieces:
+        return None
+
+    parts: list[str] = []
+    for offset, count, compressed in pieces:
+        chunk_end = min(offset + (count if compressed else count * 2), len(word_stream))
+        raw = word_stream[offset:chunk_end]
+        if not raw:
+            continue
+        if compressed:
+            parts.append(_decode_ansi(raw))
+        else:
+            parts.append(raw.decode("utf-16-le", errors="replace"))
+    return "".join(parts)
+
+
+def _parse_clx(clx: bytes) -> list[tuple[int, int, bool]]:
+    """Walk the CLX: skip Prc blocks (0x01), parse PlcPcd (0x02).
+
+    Returns ``(byte_offset_in_word_stream, char_count, is_compressed)``
+    tuples in document order.
+    """
+    pieces: list[tuple[int, int, bool]] = []
+    pos = 0
+    while pos < len(clx):
+        tag = clx[pos]
+        if tag == 0x01:  # Prc — skip its data block
+            if pos + 3 > len(clx):
+                break
+            cb_grpprl = int.from_bytes(clx[pos + 1 : pos + 3], "little")
+            pos += 3 + cb_grpprl
+        elif tag == 0x02:  # PlcPcd
+            if pos + 5 > len(clx):
+                break
+            plc_len = int.from_bytes(clx[pos + 1 : pos + 5], "little")
+            plc = clx[pos + 5 : pos + 5 + plc_len]
+            pieces.extend(_parse_plcpcd(plc))
+            break
+        else:
+            break
+    return pieces
+
+
+def _parse_plcpcd(plc: bytes) -> list[tuple[int, int, bool]]:
+    """Parse one PlcPcd: CPs[n+1] uint32 + PCDs[n] 8 bytes each.
+
+    PCD layout: 2 bytes flags, 4 bytes fc, 2 bytes prm.
+    Total PlcPcd size = 4*(n+1) + 8*n -> n = (len-4)/12, but the PCD
+    array itself is strided by 8 bytes.
+    """
+    n_pcd, rem = divmod(len(plc) - 4, 12)
+    if n_pcd <= 0 or rem != 0:
+        return []
+    cps = [
+        int.from_bytes(plc[i * 4 : i * 4 + 4], "little") for i in range(n_pcd + 1)
+    ]
+    pcds_off = (n_pcd + 1) * 4
+    pieces: list[tuple[int, int, bool]] = []
+    for i in range(n_pcd):
+        pcd = plc[pcds_off + i * 8 : pcds_off + i * 8 + 8]
+        fc = int.from_bytes(pcd[2:6], "little")
+        compressed = bool(fc & _PCD_FC_COMPRESSED)
+        real_fc = fc & 0x3FFFFFFF
+        count = cps[i + 1] - cps[i]
+        if count <= 0:
+            continue
+        pieces.append((real_fc if compressed else real_fc * 2, count, compressed))
+    return pieces
+
+
+def _extract_simple(word_stream: bytes, flags: int) -> str | None:
+    """Fallback for non-complex files: contiguous fcMin..fcMac region."""
+    fc_min = int.from_bytes(word_stream[0x18:0x1C], "little")
+    fc_mac = int.from_bytes(word_stream[0x1C:0x20], "little")
+    if not (0 <= fc_min < fc_mac <= len(word_stream)):
+        return None
+    raw = word_stream[fc_min:fc_mac]
+    if flags & _FEXTCHAR_100:
+        return raw.decode("utf-16-le", errors="replace")
+    # Old 8-bit files: CJK docs are typically GBK; latin ones cp1252.
+    try:
+        return raw.decode("gbk")
+    except UnicodeDecodeError:
+        return _decode_ansi(raw)
+
+
+def _decode_ansi(raw: bytes) -> str:
+    """Decode an 8-bit piece: prefer GBK (CJK docs), fall back to cp1252."""
+    try:
+        return raw.decode("gbk")
+    except UnicodeDecodeError:
+        return raw.decode("cp1252", errors="replace")
+
+
+def _clean(text: str) -> str:
+    """Normalize Word control characters and collapse blank runs."""
+    text = _SPECIAL_RE.sub("\n", text)
+    text = _CTRL_RE.sub("", text)
+    text = "\n".join(line.strip() for line in text.splitlines())
+    return _MULTI_BLANK_RE.sub("\n\n", text).strip()
diff --git a/app/services/note_conversion.py b/app/services/note_conversion.py
new file mode 100644
index 0000000..aee7f2d
--- /dev/null
+++ b/app/services/note_conversion.py
@@ -0,0 +1,90 @@
+"""Cloud document → note conversion (button-triggered).
+
+Orchestrates the ``note`` agent over a cloud-drive document's parsed full
+text. Layering mirrors ``services/session_summary.py``: the router passes
+the harness handle, this service does validation + lifecycle invocation and
+maps failures to HTTP status codes.
+
+The agent receives the document text as part of its input state (injected
+as a ``<document>`` block by the note graph) and saves the note itself via
+the ``save_note`` tool — so persistence, sanitization, and MySQL+Mongo
+split storage all reuse the existing note pipeline.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import HTTPException
+from loguru import logger
+
+from app.services.cloud.document_text import (
+    CloudDocumentNotFoundError,
+    read_cloud_document_text,
+)
+
+# The note agent runs a ReAct loop (organize + save_note); large documents
+# need more than the default 60s window.
+_CONVERT_TIMEOUT = 120.0
+
+_QUERY_TEMPLATE = (
+    "请把这篇云盘文档整理成一篇结构化 Markdown 笔记并保存。"
+    "文档全文见 <document> 标签，标题基于文档主题生成。"
+)
+
+
+async def convert_cloud_document_to_note(
+    uid: int,
+    upload_uuid: str,
+    agent_harness: Any,
+) -> dict[str, Any]:
+    """Convert one cloud document into a saved Markdown note.
+
+    Returns ``{"message": str}`` — the note agent's confirmation text.
+    Raises HTTPException(404/409/503/502).
+    """
+    if not (agent_harness and getattr(agent_harness, "started", False)):
+        raise HTTPException(status_code=503, detail="Agent 服务未启动")
+
+    try:
+        doc = await read_cloud_document_text(upload_uuid, uid)
+    except CloudDocumentNotFoundError:
+        raise HTTPException(
+            status_code=409,
+            detail="文档尚未解析完成或没有可用的文本内容，请先完成入库处理",
+        )
+
+    result_state = await agent_harness.lifecycle.invoke(
+        "note",
+        f"notedoc:{upload_uuid}",
+        timeout=_CONVERT_TIMEOUT,
+        uid=uid,
+        query=_QUERY_TEMPLATE,
+        cloud_upload_uuid=upload_uuid,
+        cloud_file_name=doc["file_name"],
+        cloud_doc_text=doc["content"],
+        cloud_doc_truncated=bool(doc.get("truncated")),
+    )
+    result_state = result_state or {}
+
+    error = str(result_state.get("error") or "").strip()
+    message = str(result_state.get("result") or "").strip()
+    if error or not message:
+        logger.error(
+            "[NOTE_CONVERT] failed uid=%s upload=%s error=%s",
+            uid,
+            upload_uuid[:8],
+            error or "empty result",
+        )
+        raise HTTPException(
+            status_code=502,
+            detail=f"笔记生成失败: {error or 'Agent 未返回结果'}"[:300],
+        )
+
+    logger.info(
+        "[NOTE_CONVERT] ok uid=%s upload=%s truncated=%s",
+        uid,
+        upload_uuid[:8],
+        doc.get("truncated"),
+    )
+    return {"message": message}
diff --git a/app/test/services/test_legacy_doc_preview.py b/app/test/services/test_legacy_doc_preview.py
new file mode 100644
index 0000000..6e0243e
--- /dev/null
+++ b/app/test/services/test_legacy_doc_preview.py
@@ -0,0 +1,77 @@
+"""Unit tests for legacy .doc best-effort text extraction.
+
+The full pipeline needs a real OLE2 Word file (olefile cannot write one),
+so these tests target the decoders directly with hand-crafted byte blobs,
+plus the not-a-doc rejection path.
+"""
+
+import pytest
+
+from app.services.doc_parser.legacy_doc import (
+    _clean,
+    _decode_ansi,
+    _parse_clx,
+    _parse_plcpcd,
+    extract_doc_text,
+)
+
+
+class TestDecodeAnsi:
+    def test_gbk_chinese(self):
+        raw = "工程项目报告".encode("gbk")
+        assert _decode_ansi(raw) == "工程项目报告"
+
+    def test_fallback_cp1252(self):
+        raw = b"caf\xe9"  # é in cp1252, invalid GBK sequence
+        assert _decode_ansi(raw) == "café"
+
+
+class TestParsePlcPcd:
+    def test_two_pieces_mixed_encodings(self):
+        # 2 PCDs -> 3 CPs. Piece 0: chars 0..5 compressed at fc=0x100.
+        # Piece 1: chars 5..8 UTF-16 at fc=0x200 (real byte offset 0x400).
+        cp = b"".join(n.to_bytes(4, "little") for n in (0, 5, 8))
+
+        def pcd(fc: int) -> bytes:
+            return b"\x00\x00" + fc.to_bytes(4, "little") + b"\x00\x00"
+
+        plc = cp + pcd(0x100 | 0x40000000) + pcd(0x200)
+        clx = b"\x02" + len(plc).to_bytes(4, "little") + plc
+
+        pieces = _parse_clx(clx)
+        assert pieces == [
+            (0x100, 5, True),
+            (0x400, 3, False),
+        ]
+
+    def test_malformed_lengths_rejected(self):
+        assert _parse_plcpcd(b"\x01\x02\x03") == []
+        assert _parse_plcpcd(b"") == []
+
+    def test_prc_blocks_skipped(self):
+        cp = (0).to_bytes(4, "little") + (2).to_bytes(4, "little")
+        plc = cp + b"\x00\x00" + b"\x10\x00\x00\x00" + b"\x00\x00"
+        prc = b"\x01" + (2).to_bytes(2, "little") + b"\xaa\xbb"
+        clx = prc + b"\x02" + len(plc).to_bytes(4, "little") + plc
+        pieces = _parse_clx(clx)
+        assert len(pieces) == 1
+
+
+class TestClean:
+    def test_control_chars_and_blanks(self):
+        # \r and \n each normalize to a line break -> "b\n\nc" collapses to one blank line.
+        dirty = "a\x07b\r\n\x00c\n\n\n\nd "
+        assert _clean(dirty) == "a\nb\n\nc\n\nd"
+
+    def test_empty(self):
+        assert _clean("") == ""
+
+
+class TestExtractDocText:
+    def test_non_doc_blob_returns_none(self):
+        assert extract_doc_text(b"just some text file content") is None
+        assert extract_doc_text(b"PK\x03\x04 zip bytes here") is None
+
+    @pytest.mark.parametrize("blob", [b"", b"\xec\xa5", b"\xec\xa5" + b"\x00" * 10])
+    def test_truncated_blobs_return_none(self, blob):
+        assert extract_doc_text(blob) is None
diff --git a/app/test/services/test_note_conversion.py b/app/test/services/test_note_conversion.py
new file mode 100644
index 0000000..21b04bd
--- /dev/null
+++ b/app/test/services/test_note_conversion.py
@@ -0,0 +1,180 @@
+"""Unit tests for the cloud-document → note conversion flow.
+
+Covers:
+
+* ``read_cloud_document_text`` — Mongo full-text read with ownership
+  filtering, truncation cap, and not-found semantics.
+* ``convert_cloud_document_to_note`` — orchestration: harness checks,
+  agent invocation, and error mapping to HTTP status codes.
+"""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import HTTPException
+
+from app.services.cloud.document_text import (
+    CloudDocumentNotFoundError,
+    MAX_DOC_TEXT_CHARS,
+    read_cloud_document_text,
+)
+from app.services.note_conversion import convert_cloud_document_to_note
+
+pytestmark = pytest.mark.asyncio
+
+UID = 7
+UUID = "0123abcd-4c5d-6e7f-8a9b-0123456789ab"
+
+
+# ── fakes ─────────────────────────────────────────────────────────────
+
+
+class _FakeCollection:
+    def __init__(self, doc: dict | None) -> None:
+        self._doc = doc
+        self.query: dict | None = None
+
+    async def find_one(self, query: dict, projection: dict) -> dict | None:
+        self.query = query
+        return self._doc
+
+
+class _FakeDb:
+    def __init__(self, doc: dict | None) -> None:
+        self.coll = _FakeCollection(doc)
+
+    def __getitem__(self, name: str) -> _FakeCollection:
+        return self.coll
+
+
+class _FakeLifecycle:
+    def __init__(self, result: dict | None = None, error: Exception | None = None):
+        self._result = result or {}
+        self._error = error
+        self.calls: list[dict] = []
+
+    async def invoke(self, agent_name: str, key: str, **kwargs) -> dict:
+        if self._error is not None:
+            raise self._error
+        self.calls.append({"agent": agent_name, "key": key, **kwargs})
+        return self._result
+
+
+class _FakeHarness:
+    def __init__(self, lifecycle: _FakeLifecycle, started: bool = True) -> None:
+        self.lifecycle = lifecycle
+        self.started = started
+
+
+@pytest.fixture
+def patch_mongo(monkeypatch):
+    def _install(doc: dict | None):
+        fake_db = _FakeDb(doc)
+        import app.infra.mongo as mongo_mod
+
+        monkeypatch.setattr(mongo_mod, "get_database", lambda: fake_db)
+        return fake_db
+
+    return _install
+
+
+# ── read_cloud_document_text ──────────────────────────────────────────
+
+
+class TestReadCloudDocumentText:
+    async def test_reads_full_text(self, patch_mongo):
+        patch_mongo(
+            {
+                "title": "设计文档.pdf",
+                "content": "# 标题\n正文",
+                "content_source": "pdfplumber",
+            }
+        )
+        out = await read_cloud_document_text(UUID, UID)
+
+        assert out["file_name"] == "设计文档.pdf"
+        assert out["content"] == "# 标题\n正文"
+        assert out["truncated"] is False
+
+    async def test_query_filters_by_uid(self, patch_mongo):
+        """Ownership is enforced by the (upload_uuid, uid) compound filter."""
+        fake_db = patch_mongo(None)
+
+        with pytest.raises(CloudDocumentNotFoundError):
+            await read_cloud_document_text(UUID, UID)
+        assert fake_db.coll.query == {"upload_uuid": UUID, "uid": UID}
+
+    async def test_empty_content_is_not_found(self, patch_mongo):
+        patch_mongo({"title": "x", "content": "", "content_source": ""})
+        with pytest.raises(CloudDocumentNotFoundError):
+            await read_cloud_document_text(UUID, UID)
+
+    async def test_truncation_cap(self, patch_mongo, monkeypatch):
+        long_text = "字" * (MAX_DOC_TEXT_CHARS + 1000)
+        patch_mongo({"title": "big.md", "content": long_text})
+
+        out = await read_cloud_document_text(UUID, UID)
+        assert len(out["content"]) == MAX_DOC_TEXT_CHARS
+        assert out["truncated"] is True
+
+    async def test_mongo_down_raises_runtime_error(self, monkeypatch):
+        import app.infra.mongo as mongo_mod
+
+        monkeypatch.setattr(mongo_mod, "get_database", lambda: None)
+        with pytest.raises(RuntimeError):
+            await read_cloud_document_text(UUID, UID)
+
+
+# ── convert_cloud_document_to_note ────────────────────────────────────
+
+
+class TestConvertCloudDocumentToNote:
+    async def test_success_returns_agent_message(self, patch_mongo):
+        patch_mongo(
+            {"title": "设计文档.pdf", "content": "# 标题\n正文"}
+        )
+        lifecycle = _FakeLifecycle(result={"result": "已保存笔记《设计文档》"})
+        harness = _FakeHarness(lifecycle)
+
+        out = await convert_cloud_document_to_note(UID, UUID, harness)
+
+        assert out["message"] == "已保存笔记《设计文档》"
+        call = lifecycle.calls[0]
+        assert call["agent"] == "note"
+        assert call["key"].startswith("notedoc:")
+        assert call["uid"] == UID
+        assert call["cloud_upload_uuid"] == UUID
+        assert call["cloud_file_name"] == "设计文档.pdf"
+        assert call["cloud_doc_text"] == "# 标题\n正文"
+
+    async def test_harness_not_started_is_503(self, patch_mongo):
+        with pytest.raises(HTTPException) as exc:
+            await convert_cloud_document_to_note(UID, UUID, _FakeHarness(_FakeLifecycle(), started=False))
+        assert exc.value.status_code == 503
+
+    async def test_missing_harness_is_503(self):
+        with pytest.raises(HTTPException) as exc:
+            await convert_cloud_document_to_note(UID, UUID, None)
+        assert exc.value.status_code == 503
+
+    async def test_unparsed_document_is_409(self, patch_mongo):
+        patch_mongo(None)
+        with pytest.raises(HTTPException) as exc:
+            await convert_cloud_document_to_note(
+                UID, UUID, _FakeHarness(_FakeLifecycle())
+            )
+        assert exc.value.status_code == 409
+
+    async def test_agent_error_is_502(self, patch_mongo):
+        patch_mongo({"title": "t", "content": "c"})
+        harness = _FakeHarness(_FakeLifecycle(result={"error": "boom"}))
+        with pytest.raises(HTTPException) as exc:
+            await convert_cloud_document_to_note(UID, UUID, harness)
+        assert exc.value.status_code == 502
+
+    async def test_empty_result_is_502(self, patch_mongo):
+        patch_mongo({"title": "t", "content": "c"})
+        harness = _FakeHarness(_FakeLifecycle(result={"result": ""}))
+        with pytest.raises(HTTPException) as exc:
+            await convert_cloud_document_to_note(UID, UUID, harness)
+        assert exc.value.status_code == 502